diff --git a/CHANGELOG.md b/CHANGELOG.md
index 321656d70..b30ef7bdc 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/Gemfile.lock b/Gemfile.lock
index 14944100f..2e0375ff4 100644
--- a/Gemfile.lock
+++ b/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
diff --git a/cassandra-driver.gemspec b/cassandra-driver.gemspec
index 0eda62aca..8e609f8e1 100644
--- a/cassandra-driver.gemspec
+++ b/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'
diff --git a/lib/cassandra/cluster.rb b/lib/cassandra/cluster.rb
index 7ae041034..160a9ceb2 100644
--- a/lib/cassandra/cluster.rb
+++ b/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'
diff --git a/lib/cassandra/cluster/control_connection.rb b/lib/cassandra/cluster/control_connection.rb
index 5396f2564..4932e910b 100644
--- a/lib/cassandra/cluster/control_connection.rb
+++ b/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
diff --git a/lib/cassandra/cluster/io_reactor.rb b/lib/cassandra/cluster/io_reactor.rb
new file mode 100644
index 000000000..f69d05167
--- /dev/null
+++ b/lib/cassandra/cluster/io_reactor.rb
@@ -0,0 +1,470 @@
+# encoding: utf-8
+
+#--
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#++
+
+module Cassandra
+  class Cluster
+    # An {Ione::Io::IoReactor} tuned for the way the driver uses it.
+    #
+    # It differs from the stock ione reactor in the following ways:
+    #
+    # * The TLS handshake is bounded by the connect timeout. The stock reactor
+    #   only applies the timeout to the TCP connect, so a peer that accepts TCP
+    #   but never answers the ClientHello hangs the connect forever.
+    # * A TLS socket waiting for bytes from the server is selected for
+    #   readability. The stock reactor selects every connecting socket for
+    #   writability, and a connected TCP socket is always writable, so the
+    #   reactor loop spins at 100% CPU for as long as the handshake is pending.
+    # * Closing a TLS connection also closes the underlying TCP socket, instead
+    #   of leaving the file descriptor open until garbage collection.
+    # * The reactor sleeps in select until there is IO, a timer is due, or it is
+    #   unblocked, instead of waking up every second regardless. Connect
+    #   attempts still get a bounded select so their timeouts can be enforced.
+    # * A socket whose file descriptor was closed underneath the reactor is
+    #   evicted instead of being re-selected forever.
+    # * The reactor thread is named so leaked reactors can be counted per
+    #   process (see {THREAD_NAME}).
+    #
+    # @private
+    class IoReactor < Ione::Io::IoReactor
+      # Name given to the reactor thread; count threads with this name to detect
+      # leaked reactors.
+      THREAD_NAME = 'io_reactor'.freeze
+
+      def initialize(options = {})
+        super
+        @unblocker.close
+        @unblocker = Unblocker.new
+        @scheduler = Scheduler.new(@options)
+        @io_loop   = IoLoop.new(@unblocker, @scheduler, @options)
+      end
+
+      def start
+        stopping = @lock.synchronize do
+          return @started_promise.future if @state == RUNNING_STATE
+
+          if @state == STOPPING_STATE
+            @stopped_promise.future
+          else
+            # Restore the pipe in the same transition that starts the reactor.
+            # Shutdown cannot close it between this check and a later #start.
+            if @unblocker.closed?
+              @unblocker.reopen
+              @io_loop.add_socket(@unblocker)
+            end
+            started = @started_promise = Ione::Promise.new
+            stopped = @stopped_promise = Ione::Promise.new
+            @error_listeners.each {|listener| stopped.future.on_failure(&listener)}
+            @state = RUNNING_STATE
+            ::Thread.start { run(started, stopped) }
+            return started.future
+          end
+        end
+
+        # A completed future runs its callbacks immediately, so subscribe only
+        # after releasing the state lock; restarting needs to acquire it again.
+        stopping.flat_map { start }.fallback { start }
+      end
+
+      # Same contract as {Ione::Io::IoReactor#connect}, but the connect timeout
+      # covers the TLS handshake as well as the TCP connect.
+      def connect(host, port, options = nil, &block)
+        if options.is_a?(::Numeric) || options.nil?
+          timeout = options || 5
+          ssl     = false
+        else
+          timeout = options[:timeout] || 5
+          ssl     = options[:ssl]
+        end
+
+        connection = Connection.new(host, port, timeout, @unblocker, @clock)
+        f = connection.connect
+        @io_loop.add_socket(connection)
+        @unblocker.unblock if running?
+
+        if ssl
+          f = f.flat_map do
+            ssl_context = ssl == true ? nil : ssl
+            upgraded    = SslConnection.new(host,
+                                            port,
+                                            connection.to_io,
+                                            @unblocker,
+                                            ssl_context,
+                                            connection.deadline,
+                                            @clock)
+            ff = upgraded.connect
+            @io_loop.remove_socket(connection)
+            @io_loop.add_socket(upgraded)
+            @unblocker.unblock
+            ff
+          end
+        end
+
+        f = f.map(&block) if block_given?
+        f
+      end
+
+      # The reactor sleeps until its next timer is due, so a timer scheduled
+      # while it is asleep has to wake it up.
+      def schedule_timer(timeout)
+        timer = super
+        @unblocker.unblock if running? && @io_loop.thread != ::Thread.current
+        timer
+      end
+
+      private
+
+      def run(started, stopped)
+        error = nil
+        begin
+          started.fulfill(self)
+          while @state == RUNNING_STATE
+            @io_loop.tick
+            @scheduler.tick
+          end
+        rescue => e
+          error = e
+        ensure
+          begin
+            begin
+              @io_loop.drain_sockets
+            rescue => e
+              error ||= e
+            end
+            @io_loop.close_sockets
+            @scheduler.cancel_timers
+          rescue => e
+            error ||= e
+          ensure
+            @lock.synchronize { @state = error ? CRASHED_STATE : STOPPED_STATE }
+            # A new run may start as soon as the state lock is released. Finish
+            # this run's promise, and invoke its callbacks outside the lock.
+            error ? stopped.fail(error) : stopped.fulfill(self)
+          end
+        end
+      end
+
+      # @private
+      class Unblocker < Ione::Io::Unblocker
+        # Retain the object so connections queued before a restart also use
+        # the new pipe when they need to wake the reactor.
+        def reopen
+          initialize if closed?
+        end
+      end
+
+      # @private
+      module ConnectionLifecycle
+        # Release the descriptor and wake select before notifying listeners,
+        # matching the close ordering in ione's reactor fix.
+        def close(cause = nil)
+          @lock.synchronize do
+            return false if @state == Ione::Io::BaseConnection::CLOSED_STATE
+            @state = Ione::Io::BaseConnection::CLOSED_STATE
+            @writable = false
+          end
+          if @io
+            begin
+              @io.close
+              @io = nil
+            rescue ::SystemCallError, ::IOError
+              # The descriptor may already have been closed by another thread.
+            end
+          end
+          @unblocker.unblock
+          if cause && !cause.is_a?(Ione::IoError)
+            cause = Ione::Io::ConnectionClosedError.new(cause.message)
+          end
+          cause ? @closed_promise.fail(cause) : @closed_promise.fulfill(self)
+          true
+        end
+      end
+
+      # @private
+      module AcceptorLifecycle
+        def close
+          closed = super
+          @unblocker.unblock if closed
+          closed
+        end
+
+        # The inherited drain alias bypasses an overridden close method.
+        def drain
+          close
+        end
+
+        def read
+          super
+        rescue ::IOError, ::Errno::EBADF
+          close
+        rescue ::SystemCallError
+          # A transient accept error leaves the listener available for retry.
+          nil
+        end
+      end
+
+      # @private
+      class Connection < Ione::Io::Connection
+        include ConnectionLifecycle
+
+        attr_reader :deadline
+
+        def initialize(*args)
+          super
+          # Time cannot represent infinity; nil leaves the connect unbounded.
+          @deadline = @clock.now + @connection_timeout unless @connection_timeout == ::Float::INFINITY
+        end
+
+        def connect
+          return @connected_promise.future if closed?
+
+          # Let a completed kernel handshake win over the deadline. The parent
+          # leaves the connection pending only after EINPROGRESS or EALREADY.
+          future = super
+          if @deadline && connecting? && @clock.now >= @deadline
+            close(Ione::Io::ConnectionTimeoutError.new(
+                    "Could not connect to #{@host}:#{@port} within #{@connection_timeout}s"
+            ))
+          end
+          future
+        rescue ::IOError => e
+          close(e)
+          @connected_promise.future
+        end
+      end
+
+      # @private
+      class Scheduler < Ione::Io::Scheduler
+        def initialize(options = {})
+          super
+          @clock = options[:clock] || ::Time
+        end
+
+        # @return [Numeric, nil] seconds until the earliest pending timer is
+        #   due (zero when overdue), or nil when there are no timers
+        def next_timeout
+          timer = @lock.synchronize { @timer_queue.peek }
+          return nil unless timer
+
+          remaining = timer.time - @clock.now
+          remaining > 0 ? remaining : 0
+        end
+      end
+
+      # @private
+      class IoLoop < Ione::Io::IoLoopBody
+        attr_reader :thread
+
+        def initialize(unblocker, scheduler, options = {})
+          super(unblocker, options)
+          @scheduler       = scheduler
+          @clock           = options[:clock] || ::Time
+          @tick_resolution = options[:tick_resolution] || 1
+          @drain_timeout   = options[:drain_timeout] || 5
+        end
+
+        def add_socket(socket)
+          # bind/accept create ione's server sockets. Apply the same lifecycle
+          # fixes to those instances without modifying the dependency globally.
+          if socket.is_a?(Ione::Io::ServerConnection)
+            socket.extend(ConnectionLifecycle)
+          elsif socket.is_a?(Ione::Io::Acceptor)
+            socket.extend(AcceptorLifecycle)
+          end
+          super
+        end
+
+        # @param timeout [Numeric, nil] fixed select timeout while draining;
+        #   otherwise derived from pending timers and connect deadlines
+        def tick(timeout = nil)
+          name_thread
+
+          readables  = []
+          writables  = []
+          connecting = []
+
+          @sockets.each do |s|
+            if s.connected?
+              readables << s
+            elsif s.connecting?
+              connecting << s
+              # A TLS handshake waiting on the server must be selected for
+              # readability; selecting it for writability returns immediately
+              # and spins the loop.
+              if s.respond_to?(:handshake_wants_read?) && s.handshake_wants_read?
+                readables << s
+                next
+              end
+            end
+
+            writables << s if s.connecting? || s.writable?
+          end
+
+          unless timeout
+            deadlines = connecting.map do |s|
+              if s.respond_to?(:deadline)
+                deadline = s.deadline
+                [deadline - @clock.now, 0].max if deadline
+              else
+                @tick_resolution
+              end
+            end
+            timeout = [@scheduler.next_timeout, *deadlines].compact.min
+          end
+
+          begin
+            r, w, _ = @selector.select(readables, writables, nil, timeout)
+          rescue ::IOError, ::Errno::EBADF, ::TypeError => e
+            raise unless evict_dead_sockets(readables + writables, e)
+            return
+          end
+          connecting.each {|s| dispatch(s, :connect)}
+          r && r.each {|s| dispatch(s, :read) if s.connected?}
+          w && w.each {|s| dispatch(s, :flush)}
+        end
+
+        def drain_sockets
+          threshold = @clock.now + @drain_timeout
+          until @clock.now >= threshold || @sockets.none?(&:writable?)
+            @sockets.each(&:drain)
+            tick(@tick_resolution)
+            @lock.synchronize { @sockets = @sockets.reject(&:closed?) }
+          end
+          if @clock.now >= threshold
+            raise Ione::Io::ReactorError,
+                  format('Socket drain timeout after %p s', @drain_timeout)
+          end
+        end
+
+        private
+
+        def name_thread
+          @thread = ::Thread.current
+          @thread.name = THREAD_NAME if @thread.name.nil?
+        end
+
+        def dispatch(socket, method)
+          socket.__send__(method)
+        rescue ::IOError, ::Errno::EBADF => e
+          close_socket(socket, e)
+        end
+
+        def close_socket(socket, error)
+          socket.is_a?(Ione::Io::BaseConnection) ? socket.close(error) : socket.close
+        rescue
+          # The descriptor may already be closed.
+          nil
+        end
+
+        # select raised because a file descriptor in the set is closed. Close
+        # and drop the offending sockets so the loop cannot spin on them.
+        def evict_dead_sockets(sockets, error)
+          dead = sockets.uniq.select do |s|
+            next true if s.closed?
+            begin
+              io = s.to_io
+              next true if io.nil? || io.closed?
+              # A descriptor closed through another IO wrapper can still
+              # report closed? == false. Probe the actual descriptor.
+              ::IO.select([io], nil, nil, 0)
+              false
+            rescue ::IOError, ::Errno::EBADF, ::TypeError
+              true
+            end
+          end
+          dead.each {|s| close_socket(s, error)}
+          @lock.synchronize { @sockets = @sockets.reject {|s| s.closed? || dead.include?(s)} }
+          !dead.empty?
+        end
+      end
+
+      # @private
+      class SslConnection < Ione::Io::SslConnection
+        include ConnectionLifecycle
+
+        attr_reader :deadline
+
+        def initialize(host, port, io, unblocker, ssl_context, deadline, clock)
+          super(host, port, io, unblocker, ssl_context)
+          @deadline   = deadline
+          @clock      = clock
+          @wants_read = false
+        end
+
+        # @return [Boolean] true while the handshake is waiting for bytes from
+        #   the server
+        def handshake_wants_read?
+          @wants_read
+        end
+
+        def connect
+          return @connected_promise.future if closed? || connected?
+
+          # Check the deadline only if the handshake still needs IO.
+          if @io.nil?
+            @io = if @ssl_context
+                    @socket_impl.new(@raw_io, @ssl_context)
+                  else
+                    @socket_impl.new(@raw_io)
+                  end
+            @io.sync_close = true if @io.respond_to?(:sync_close=)
+          end
+          @io.connect_nonblock
+          @wants_read = false
+          @state = CONNECTED_STATE
+          @connected_promise.fulfill(self)
+          @connected_promise.future
+        rescue ::IO::WaitReadable
+          @wants_read = true
+          fail_if_past_deadline
+          @connected_promise.future
+        rescue ::IO::WaitWritable
+          @wants_read = false
+          fail_if_past_deadline
+          @connected_promise.future
+        rescue => e
+          close(e)
+          @connected_promise.future
+        end
+
+        def close(cause = nil)
+          closed = super
+          if closed
+            # The SSL socket closes the raw socket when it exists (sync_close),
+            # but a handshake that never started leaves only the raw socket.
+            begin
+              @raw_io.close if @raw_io && !@raw_io.closed?
+            rescue ::SystemCallError, ::IOError
+              nil
+            end
+          end
+          closed
+        end
+
+        private
+
+        def fail_if_past_deadline
+          return if @deadline.nil? || @clock.now < @deadline
+          close(Ione::Io::ConnectionTimeoutError.new(
+                  "Could not complete TLS handshake with #{@host}:#{@port} " \
+                  'within the connect timeout'
+          ))
+        end
+      end
+    end
+  end
+end
diff --git a/lib/cassandra/driver.rb b/lib/cassandra/driver.rb
index f99df3e34..642122935 100644
--- a/lib/cassandra/driver.rb
+++ b/lib/cassandra/driver.rb
@@ -28,7 +28,7 @@ def self.let(name, &block)
       define_method(:"#{name}=") { |object| @instances[name] = object }
     end
 
-    let(:io_reactor)       { Ione::Io::IoReactor.new }
+    let(:io_reactor)       { Cluster::IoReactor.new }
     let(:cluster_registry) { Cluster::Registry.new(logger) }
     let(:cluster_schema)   { Cluster::Schema.new }
     let(:cluster_metadata) do
diff --git a/lib/cassandra/reconnection/policies/exponential.rb b/lib/cassandra/reconnection/policies/exponential.rb
index ffd84fb84..98740fe80 100644
--- a/lib/cassandra/reconnection/policies/exponential.rb
+++ b/lib/cassandra/reconnection/policies/exponential.rb
@@ -20,18 +20,23 @@ module Cassandra
   module Reconnection
     module Policies
       # A reconnection policy that returns a constant exponentially growing
-      # reconnection interval up to a given maximum
+      # reconnection interval up to a given maximum, optionally randomised by
+      # a jitter fraction so that many clients do not reconnect in lockstep.
       class Exponential < Policy
         # @private
         class Schedule
-          def initialize(start, max, exponent)
+          def initialize(start, max, exponent, jitter, random)
             @interval = start
             @max      = max
             @exponent = exponent
+            @jitter   = jitter
+            @random   = random
           end
 
           def next
-            @interval.tap { backoff if @interval < @max }
+            interval = @interval
+            backoff if @interval < @max
+            randomize(interval)
           end
 
           private
@@ -45,11 +50,27 @@ def backoff
                           new_interval
                         end
           end
+
+          # Spreads the interval by up to +/- jitter, never above the maximum.
+          def randomize(interval)
+            return interval if @jitter.zero?
+
+            lower = interval * (1 - @jitter)
+            upper = [interval * (1 + @jitter), @max].min
+            lower + (upper - lower) * @random.rand
+          end
         end
 
         # @param start    [Numeric] beginning interval
-        # @param max      [Numeric] maximum reconnection interval
+        # @param max      [Numeric] maximum reconnection interval; never
+        #   exceeded, even with jitter
         # @param exponent [Numeric] (2) interval exponent to use
+        # @param jitter   [Numeric] (0) fraction in `0...1` by which each
+        #   interval is randomised, e.g. `0.25` for +/- 25%
+        # @param random   [#rand] (Random) source of randomness for
+        #   jitter
+        #
+        # @raise [ArgumentError] if jitter is not in `0...1`
         #
         # @example Using this policy
         #   policy   = Cassandra::Reconnection::Policies::Exponential.new(0.5, 10, 2)
@@ -62,16 +83,35 @@ def backoff
         #   schedule.next # 10.0
         #   schedule.next # 10.0
         #   schedule.next # 10.0
-        def initialize(start, max, exponent = 2)
+        #
+        # @example With jitter, so that a fleet does not reconnect in lockstep
+        #   policy   = Cassandra::Reconnection::Policies::Exponential.new(1, 60, 2, jitter: 0.25)
+        #   schedule = policy.schedule
+        #   schedule.next # somewhere in 0.75..1.25
+        #   schedule.next # somewhere in 1.5..2.5
+        #   # ...
+        #   schedule.next # somewhere in 45..60
+        def initialize(start, max, exponent = 2, jitter: 0, random: ::Random)
+          begin
+            jitter = Float(jitter)
+          rescue ::TypeError, ::ArgumentError
+            raise ::ArgumentError, "jitter must be in 0...1, #{jitter.inspect} given"
+          end
+          unless jitter >= 0 && jitter < 1
+            raise ::ArgumentError, "jitter must be in 0...1, #{jitter.inspect} given"
+          end
+
           @start    = start
           @max      = max
           @exponent = exponent
+          @jitter   = jitter
+          @random   = random
         end
 
         # @return [Cassandra::Reconnection::Schedule] an exponential
         #   reconnection schedule
         def schedule
-          Schedule.new(@start, @max, @exponent)
+          Schedule.new(@start, @max, @exponent, @jitter, @random)
         end
       end
     end
diff --git a/lib/cassandra/version.rb b/lib/cassandra/version.rb
index ed8d2183c..714ee21e5 100644
--- a/lib/cassandra/version.rb
+++ b/lib/cassandra/version.rb
@@ -17,5 +17,5 @@
 #++
 
 module Cassandra
-  VERSION = '3.2.5'.freeze
+  VERSION = '3.2.6'.freeze
 end
diff --git a/spec/cassandra/cluster/control_connection_spec.rb b/spec/cassandra/cluster/control_connection_spec.rb
index aeecb4c51..966ae2f5e 100644
--- a/spec/cassandra/cluster/control_connection_spec.rb
+++ b/spec/cassandra/cluster/control_connection_spec.rb
@@ -1059,6 +1059,62 @@ def handle_request(&handler)
       end
 
       describe "#close_async" do
+        it 'stops the reactor even before the first connect' do
+          expect(io_reactor).to receive(:stop).once.and_return(Ione::Future.resolved)
+          expect(control_connection.close_async).to be_resolved
+          expect(control_connection.close_async).to be_resolved
+        end
+
+        it 'does not connect on a host-up notification before the first connect' do
+          expect(io_reactor).not_to receive(:connect)
+          control_connection.host_up(cluster_registry.hosts.first).value
+        end
+
+        it 'can close again after an explicit reconnect' do
+          control_connection.connect_async.value
+          first_close = control_connection.close_async
+          expect(first_close).to be_resolved
+
+          control_connection.connect_async.value
+          stopped = Ione::Promise.new
+          expect(io_reactor).to receive(:stop).once.and_return(stopped.future)
+          second_close = control_connection.close_async
+          expect(second_close).not_to equal(first_close)
+          expect(second_close).not_to be_completed
+          stopped.fulfill
+          expect(second_close).to be_resolved
+        end
+
+        it 'keeps the original close future when a listener reconnects' do
+          control_connection.connect_async.value
+          first_close = nil
+          control_connection.on_close do
+            first_close = control_connection.close_async
+            control_connection.connect_async.value
+          end
+          returned = control_connection.close_async
+
+          expect(returned).to equal(first_close)
+          expect(first_close).to be_resolved
+          expect(io_reactor).to be_running
+        end
+
+        it 'stops only once when a socket closes before the stop future resolves' do
+          control_connection.connect_async.value
+          stopped = Ione::Promise.new
+          expect(io_reactor).to receive(:stop).once do
+            last_connection.close
+            stopped.future
+          end
+
+          closing = control_connection.close_async
+          expect(control_connection.close_async).to equal(closing)
+          expect(closing).not_to be_completed
+          expect { control_connection.connect_async.value }.to raise_error(Errors::ClientError, /closing/)
+          stopped.fulfill
+          expect(closing).to be_resolved
+        end
+
         context 'when connected' do
           before do
             control_connection.connect_async.value
diff --git a/spec/cassandra/cluster/io_reactor_spec.rb b/spec/cassandra/cluster/io_reactor_spec.rb
new file mode 100644
index 000000000..71908e19f
--- /dev/null
+++ b/spec/cassandra/cluster/io_reactor_spec.rb
@@ -0,0 +1,799 @@
+# encoding: utf-8
+
+#--
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#++
+
+require 'spec_helper'
+require 'socket'
+require 'openssl'
+require 'timeout'
+
+# These specs only ever talk to loopback sockets opened by the spec itself.
+module Cassandra
+  class Cluster
+    describe(IoReactor) do
+      # Wraps IO.select so the spec can see how often, and for how long, the
+      # reactor loop goes to sleep.
+      class CountingSelector
+        attr_reader :calls, :timeouts
+
+        def initialize
+          @calls    = 0
+          @timeouts = []
+        end
+
+        def select(readables, writables, errorables, timeout)
+          @calls += 1
+          @timeouts << timeout
+          ::IO.select(readables, writables, errorables, timeout)
+        end
+      end
+
+      let(:selector) { CountingSelector.new }
+      let(:reactor)  { IoReactor.new(selector: selector) }
+
+      after do
+        begin
+          ::Timeout.timeout(3) { reactor.stop.value }
+        rescue
+          reactor.instance_variable_get(:@io_loop).thread.kill if reactor.running?
+        end
+      end
+
+      def free_port
+        server = ::TCPServer.new('127.0.0.1', 0)
+        port   = server.addr[1]
+        server.close
+        port
+      end
+
+      def reactor_threads
+        ::Thread.list.select {|t| t.name == IoReactor::THREAD_NAME}
+      end
+
+      describe('thread') do
+        it 'is named io_reactor while running and exits on stop' do
+          before = ::Thread.list.size
+
+          reactor.start.value
+          await { reactor_threads.size == 1 }
+
+          reactor.stop.value
+          await { ::Thread.list.size <= before }
+          expect(reactor_threads).to be_empty
+        end
+
+        it 'wakes for timers and stops after repeated restarts' do
+          ::Timeout.timeout(3) do
+            3.times do
+              reactor.start.value
+              reactor.schedule_timer(0.01).value
+              reactor.stop.value
+            end
+          end
+          expect(reactor_threads).to be_empty
+        end
+
+        it 'restarts when start is requested while stopping' do
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            restarted = reactor.schedule_timer(0).flat_map do
+              reactor.stop
+              reactor.start
+            end
+            expect(restarted.value).to eq(reactor)
+            reactor.schedule_timer(0.01).value
+            reactor.stop.value
+          end
+        end
+
+        it 'restarts when shutdown completes immediately after start observes stopping' do
+          entered_select = ::Queue.new
+          release_select = ::Queue.new
+          observed_stopping = ::Queue.new
+          resume_start = ::Queue.new
+          begin_start = ::Queue.new
+          requester = nil
+          first_select = true
+          paused_start = false
+
+          allow(selector).to receive(:select).and_wrap_original do |original, *args|
+            if first_select
+              first_select = false
+              entered_select << true
+              release_select.pop
+              nil
+            else
+              original.call(*args)
+            end
+          end
+
+          lock = reactor.instance_variable_get(:@lock)
+          allow(lock).to receive(:synchronize).and_wrap_original do |original, &block|
+            result = original.call(&block)
+            if ::Thread.current == requester && !paused_start
+              paused_start = true
+              observed_stopping << true
+              resume_start.pop
+            end
+            result
+          end
+
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            entered_select.pop
+            stopped = reactor.stop
+            requester = ::Thread.new do
+              begin_start.pop
+              reactor.start
+            end
+            begin_start << true
+
+            # Let shutdown close the pipe after start releases the state lock,
+            # before it can continue starting or subscribe to the stop future.
+            observed_stopping.pop
+            release_select << true
+            stopped.value
+            resume_start << true
+
+            expect(requester.value.value).to eq(reactor)
+            reactor.schedule_timer(0.01).value
+            reactor.stop.value
+          end
+        ensure
+          release_select << true
+          resume_start << true
+          begin_start << true
+          requester.kill.join if requester && requester.alive?
+        end
+
+        it 'restores the unblocker after a reactor crash' do
+          crashed = Ione::Promise.new
+          reactor.on_error {|error| crashed.fulfill(error)}
+          allow(selector).to receive(:select).and_wrap_original do |original, *args|
+            if crashed.future.completed?
+              original.call(*args)
+            else
+              raise 'selector failure'
+            end
+          end
+
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            expect(crashed.future.value.message).to eq('selector failure')
+            reactor.start.value
+            reactor.schedule_timer(0.01).value
+            reactor.stop.value
+          end
+        end
+
+        it 'completes the old stop future when a new run has already started' do
+          shutdown_finished = ::Queue.new
+          complete_stop = ::Queue.new
+          paused_completion = false
+          previous_thread = nil
+
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            await { reactor.instance_variable_get(:@io_loop).thread }
+            previous_thread = reactor.instance_variable_get(:@io_loop).thread
+            lock = reactor.instance_variable_get(:@lock)
+            allow(lock).to receive(:synchronize).and_wrap_original do |original, &block|
+              result = original.call(&block)
+              if ::Thread.current == previous_thread &&
+                 reactor.instance_variable_get(:@state) == IoReactor::STOPPED_STATE &&
+                 !paused_completion
+                paused_completion = true
+                shutdown_finished << true
+                complete_stop.pop
+              end
+              result
+            end
+
+            old_stop = reactor.stop
+            shutdown_finished.pop
+            reactor.start.value
+            new_stop = reactor.instance_variable_get(:@stopped_promise).future
+            expect(old_stop).not_to be_completed
+            complete_stop << true
+
+            expect(old_stop.value).to eq(reactor)
+            expect(new_stop).not_to be_completed
+            reactor.schedule_timer(0.01).value
+            expect(reactor.stop).to equal(new_stop)
+            new_stop.value
+          end
+        ensure
+          complete_stop << true
+          previous_thread.join(1) if previous_thread
+        end
+      end
+
+      describe('when idle') do
+        it 'sleeps in select without a timeout instead of ticking' do
+          reactor.start.value
+          sleep(0.3)
+
+          # start plus draining the unblocker; a ticking reactor would have
+          # woken up a few hundred times by now with a small tick resolution
+          expect(selector.calls).to be <= 3
+          expect(selector.timeouts.last).to be_nil
+        end
+
+        it 'wakes up for a timer scheduled while it is asleep' do
+          reactor.start.value
+          sleep(0.1)
+
+          started = ::Time.now
+          reactor.schedule_timer(0.05).value
+          expect(::Time.now - started).to be < 1
+        end
+
+        it 'sleeps only until the next timer is due' do
+          reactor.start.value
+          sleep(0.1)
+
+          reactor.schedule_timer(0.2)
+          await { selector.timeouts.last && selector.timeouts.last <= 0.2 }
+          expect(selector.timeouts.last).to be > 0
+        end
+
+        it 'does not wake itself when scheduling a timer on the reactor thread' do
+          reactor.start.value
+          await { selector.calls > 0 }
+          unblocker = reactor.instance_variable_get(:@unblocker)
+          allow(unblocker).to receive(:unblock).and_call_original
+
+          ::Timeout.timeout(3) do
+            reactor.schedule_timer(0).flat_map { reactor.schedule_timer(0.01) }.value
+          end
+
+          expect(unblocker).to have_received(:unblock).once
+        end
+
+        [:close, :drain].each do |operation|
+          it "wakes when an outgoing connection receives #{operation} from another thread" do
+            server = ::TCPServer.new('127.0.0.1', 0)
+            reactor.start.value
+            connection = reactor.connect('127.0.0.1', server.addr[1], 1).value
+            accepted = server.accept
+            unblocker = reactor.instance_variable_get(:@unblocker)
+            events = []
+            allow(unblocker).to receive(:unblock).and_wrap_original do |original|
+              events << :wakeup
+              original.call
+            end
+            connection.on_closed { events << :closed }
+            sleep(0.1)
+            calls = selector.calls
+
+            connection.public_send(operation)
+
+            expect(unblocker).to have_received(:unblock)
+            expect(events.index(:wakeup)).to be < events.index(:closed)
+            await { selector.calls > calls }
+            expect(::Timeout.timeout(2) { accepted.read }).to eq('')
+          ensure
+            accepted.close if accepted && !accepted.closed?
+            server.close if server && !server.closed?
+          end
+
+          it "releases a listening port when its acceptor receives #{operation}" do
+            reactor.start.value
+            acceptor = reactor.bind('127.0.0.1', 0, 5).value
+            port = acceptor.to_io.local_address.ip_port
+            unblocker = reactor.instance_variable_get(:@unblocker)
+            allow(unblocker).to receive(:unblock).and_call_original
+            sleep(0.1)
+            calls = selector.calls
+
+            acceptor.public_send(operation)
+
+            expect(unblocker).to have_received(:unblock)
+            await { selector.calls > calls }
+            rebound = ::TCPServer.new('127.0.0.1', port)
+          ensure
+            rebound.close if rebound
+          end
+
+          it "wakes when an accepted connection receives #{operation} from another thread" do
+            reactor.start.value
+            acceptor = reactor.bind('127.0.0.1', 0, 5).value
+            accepted = Ione::Promise.new
+            acceptor.on_accept {|connection| accepted.fulfill(connection)}
+            peer = ::TCPSocket.new('127.0.0.1', acceptor.to_io.local_address.ip_port)
+            connection = ::Timeout.timeout(2) { accepted.future.value }
+            unblocker = reactor.instance_variable_get(:@unblocker)
+            allow(unblocker).to receive(:unblock).and_call_original
+            sleep(0.1)
+            calls = selector.calls
+
+            connection.public_send(operation)
+
+            expect(unblocker).to have_received(:unblock)
+            await { selector.calls > calls }
+            expect(::Timeout.timeout(2) { peer.read }).to eq('')
+          ensure
+            peer.close if peer && !peer.closed?
+          end
+        end
+      end
+
+      describe('TLS handshake') do
+        let(:server) { ::TCPServer.new('127.0.0.1', 0) }
+        let(:port)   { server.addr[1] }
+
+        after { server.close unless server.closed? }
+
+        context('when the server accepts TCP but never answers the handshake') do
+          it 'fails within the connect timeout instead of hanging' do
+            reactor.start.value
+
+            started  = ::Time.now
+            future   = reactor.connect('127.0.0.1', port, timeout: 0.5, ssl: true)
+            accepted = server.accept
+
+            expect { ::Timeout.timeout(3) { future.value } }.to raise_error(Ione::Io::ConnectionTimeoutError, /TLS handshake/)
+            expect(::Time.now - started).to be < 1
+
+            # the half-open socket is closed, not left in the loop: reading
+            # drains the ClientHello and then hits EOF instead of blocking
+            expect { ::Timeout.timeout(2) { accepted.read } }.not_to raise_error
+            accepted.close
+          end
+
+          it 'does not spin the reactor loop while the handshake is pending' do
+            reactor.start.value
+
+            future = reactor.connect('127.0.0.1', port, timeout: 1, ssl: true)
+            accepted = server.accept
+            begin
+              ::Timeout.timeout(3) { future.value }
+            rescue Ione::Io::ConnectionTimeoutError
+              nil
+            end
+            accepted.close
+
+            # the stock reactor selects the socket for writability, which
+            # returns immediately, and goes round tens of thousands of times
+            expect(selector.calls).to be < 20
+          end
+
+          it 'keeps an infinite timeout pending while allowing timers and shutdown' do
+            accepted = nil
+            ::Timeout.timeout(3) do
+              reactor.start.value
+              future = reactor.connect('127.0.0.1', port, timeout: Float::INFINITY, ssl: true)
+              accepted = server.accept
+
+              reactor.schedule_timer(0.05).value
+              expect(future).not_to be_completed
+              expect(selector.calls).to be < 20
+
+              reactor.stop.value
+              expect(future).to be_completed
+              expect(accepted.read).not_to be_empty
+            end
+          ensure
+            accepted.close if accepted
+          end
+        end
+
+        context('against a local TLS server') do
+          let(:key) { ::OpenSSL::PKey::RSA.new(2048) }
+          let(:cert) do
+            cert = ::OpenSSL::X509::Certificate.new
+            cert.version    = 2
+            cert.serial     = 1
+            cert.subject    = ::OpenSSL::X509::Name.parse('/CN=127.0.0.1')
+            cert.issuer     = cert.subject
+            cert.public_key = key.public_key
+            cert.not_before = ::Time.now - 60
+            cert.not_after  = ::Time.now + 3600
+            cert.sign(key, ::OpenSSL::Digest::SHA256.new)
+            cert
+          end
+          let(:server_context) do
+            ctx      = ::OpenSSL::SSL::SSLContext.new
+            ctx.cert = cert
+            ctx.key  = key
+            ctx
+          end
+          let(:client_context) do
+            ctx             = ::OpenSSL::SSL::SSLContext.new
+            ctx.verify_mode = ::OpenSSL::SSL::VERIFY_NONE
+            ctx
+          end
+
+          it 'completes the handshake and receives data' do
+            ssl_server = ::OpenSSL::SSL::SSLServer.new(server, server_context)
+            server_thread = ::Thread.new do
+              socket = ssl_server.accept
+              socket.write('hello')
+              socket.flush
+              sleep(0.2)
+              socket.close
+            end
+
+            reactor.start.value
+            connection = reactor.connect('127.0.0.1', port, timeout: 5, ssl: client_context).value
+            expect(connection).to be_connected
+
+            received = +''
+            connection.on_data {|data| received << data}
+            await { received == 'hello' }
+
+            server_thread.join(5)
+          end
+        end
+      end
+
+      describe('accept errors') do
+        [::IOError, ::Errno::EBADF, ::Errno::ECONNABORTED].each do |error_class|
+          it "handles #{error_class} at the listener without stopping the reactor" do
+            reactor.start.value
+            acceptor = reactor.bind('127.0.0.1', 0, 5).value
+            allow(acceptor.to_io).to receive(:accept_nonblock).and_raise(error_class)
+
+            expect { acceptor.read }.not_to raise_error
+
+            expect(acceptor.closed?).to eq(error_class != ::Errno::ECONNABORTED)
+            ::Timeout.timeout(2) { reactor.schedule_timer(0.01).value }
+            expect(reactor).to be_running
+          end
+        end
+      end
+
+      describe('plain TCP') do
+        context('when TCP completes while the reactor is stopped') do
+          let(:clock) { double('clock', now: 100.0) }
+          let(:reactor) { IoReactor.new(selector: selector, clock: clock) }
+
+          it 'uses the completed connection even when the deadline has elapsed before restart' do
+            server = ::TCPServer.new('127.0.0.1', 0)
+            reactor.start.value
+            reactor.stop.value
+            connected = reactor.connect('127.0.0.1', server.addr[1], timeout: 0.25)
+            peer = server.accept
+            allow(clock).to receive(:now).and_return(100.25)
+
+            ::Timeout.timeout(3) do
+              reactor.start.value
+              connected.value.write('hello')
+              expect(peer.read(5)).to eq('hello')
+              reactor.stop.value
+            end
+          ensure
+            peer.close if peer
+            server.close if server
+          end
+        end
+
+        it 'fails fast when nothing is listening' do
+          reactor.start.value
+
+          started = ::Time.now
+          future  = reactor.connect('127.0.0.1', free_port, timeout: 5)
+          expect { future.value }.to raise_error(Ione::Io::ConnectionError)
+          expect(::Time.now - started).to be < 2
+        end
+
+        it 'can write through a connection queued before restarting' do
+          server = ::TCPServer.new('127.0.0.1', 0)
+          reactor.start.value
+          reactor.stop.value
+          connected = reactor.connect('127.0.0.1', server.addr[1], timeout: 1)
+          accepted = server.accept
+
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            connected.value.write('hello')
+            expect(accepted.read(5)).to eq('hello')
+            reactor.stop.value
+          end
+        ensure
+          accepted.close if accepted
+          server.close if server
+        end
+
+        it 'connects and writes with an infinite timeout' do
+          accepted = nil
+          server = ::TCPServer.new('127.0.0.1', 0)
+          ::Timeout.timeout(3) do
+            reactor.start.value
+            connection = reactor.connect('127.0.0.1', server.addr[1], Float::INFINITY).value
+            accepted = server.accept
+            connection.write('hello')
+            expect(accepted.read(5)).to eq('hello')
+          end
+        ensure
+          accepted.close if accepted
+          server.close if server
+        end
+      end
+    end
+
+    describe(IoReactor::IoLoop) do
+      let(:clock) { double('clock', now: 100.0) }
+      let(:selector) { double('selector') }
+      let(:unblocker) { IoReactor::Unblocker.new }
+      let(:scheduler) { IoReactor::Scheduler.new(clock: clock) }
+      let(:io_loop) do
+        IoReactor::IoLoop.new(unblocker, scheduler,
+                             clock: clock, selector: selector, tick_resolution: 1, drain_timeout: 3)
+      end
+
+      before { @thread_name = ::Thread.current.name }
+      after do
+        io_loop.close_sockets
+        ::Thread.current.name = @thread_name
+      end
+
+      it 'uses a fixed timeout while draining with an overdue timer' do
+        timer = scheduler.schedule_timer(-1)
+        socket = double('socket', connected?: false, connecting?: false,
+                                  writable?: true, closed?: false, drain: nil, close: nil)
+        io_loop.add_socket(socket)
+        now = 100.0
+        allow(selector).to receive(:select) do |_, _, _, timeout|
+          expect(timeout).to eq(1)
+          now += timeout
+          allow(clock).to receive(:now).and_return(now)
+          nil
+        end
+
+        expect { io_loop.drain_sockets }.to raise_error(Ione::Io::ReactorError, /drain timeout/)
+        expect(selector).to have_received(:select).exactly(3).times
+        expect(timer).not_to be_completed
+      end
+
+      it 'selects only until the nearest connect deadline' do
+        [100.25, nil, 102.0].each do |deadline|
+          socket = double('connection', connected?: false, connecting?: true,
+                                        closed?: false, deadline: deadline, connect: nil, close: nil)
+          io_loop.add_socket(socket)
+        end
+        scheduler.schedule_timer(0.5)
+
+        expect(selector).to receive(:select).with(anything, anything, nil, 0.25)
+        io_loop.tick
+      end
+
+      it 'selects without a deadline for an infinite timeout' do
+        socket = double('connection', connected?: false, connecting?: true,
+                                      closed?: false, deadline: nil, connect: nil, close: nil)
+        io_loop.add_socket(socket)
+
+        expect(selector).to receive(:select).with(anything, anything, nil, nil)
+        io_loop.tick
+      end
+
+      it 'honours timers while a connection has an infinite timeout' do
+        socket = double('connection', connected?: false, connecting?: true,
+                                      closed?: false, deadline: nil, connect: nil, close: nil)
+        io_loop.add_socket(socket)
+        scheduler.schedule_timer(0.25)
+
+        expect(selector).to receive(:select).with(anything, anything, nil, 0.25)
+        io_loop.tick
+      end
+
+      it 'lets an earlier timer bound select while connecting' do
+        socket = double('connection', connected?: false, connecting?: true,
+                                      closed?: false, deadline: 105.0, connect: nil, close: nil)
+        io_loop.add_socket(socket)
+        scheduler.schedule_timer(0.1)
+
+        expect(selector).to receive(:select) do |_, _, _, timeout|
+          expect(timeout).to be_within(0.0001).of(0.1)
+          nil
+        end
+        io_loop.tick
+      end
+
+      def add_connection(io)
+        connection = Ione::Io::BaseConnection.new('127.0.0.1', 9042, unblocker)
+        connection.instance_variable_set(:@io, io)
+        connection.instance_variable_set(:@state, Ione::Io::BaseConnection::CONNECTED_STATE)
+        io_loop.add_socket(connection)
+        connection
+      end
+
+      [:connect, :read, :flush].each do |operation|
+        [::IOError, ::Errno::EBADF].each do |error_class|
+          it "isolates #{error_class} during #{operation} and continues reading healthy connections" do
+            reader, writer = ::IO.pipe
+            bad = add_connection(reader.dup)
+            healthy = add_connection(reader)
+            if operation == :connect
+              bad.instance_variable_set(:@state, Ione::Io::BaseConnection::CONNECTING_STATE)
+            elsif operation == :flush
+              allow(bad).to receive(:writable?).and_return(true)
+            end
+            allow(bad).to receive(operation).and_raise(error_class, 'closed stream')
+            selected_readers = operation == :read ? [bad, healthy] : [healthy]
+            selected_writers = operation == :flush ? [bad] : nil
+            allow(selector).to receive(:select).and_return([selected_readers, selected_writers, nil])
+            received = nil
+            healthy.on_data {|data| received = data}
+            writer.write('hello')
+
+            expect { io_loop.tick }.not_to raise_error
+
+            expect(bad).to be_closed
+            expect(healthy).not_to be_closed
+            expect(received).to eq('hello')
+          ensure
+            writer.close if writer
+          end
+        end
+      end
+
+      it 'propagates dispatch errors unrelated to closed sockets' do
+        reader, writer = ::IO.pipe
+        connection = add_connection(reader)
+        allow(connection).to receive(:read).and_raise(ArgumentError, 'invalid handler')
+        allow(selector).to receive(:select).and_return([[connection], nil, nil])
+
+        expect { io_loop.tick }.to raise_error(ArgumentError, 'invalid handler')
+      ensure
+        writer.close if writer
+      end
+
+      [::IOError, ::Errno::EBADF, ::TypeError].each do |error_class|
+        it "evicts an invalid socket after #{error_class} and retains healthy sockets" do
+          reader, writer = ::IO.pipe
+          dead_io = reader.dup
+          dead = add_connection(dead_io)
+          healthy = add_connection(reader)
+          if error_class == ::IOError
+            dead_io.close
+          elsif error_class == ::Errno::EBADF
+            # Close the fd through another wrapper: closed? still returns false.
+            ::IO.for_fd(dead_io.fileno).close
+            expect(dead_io).not_to be_closed
+          else
+            dead_io.close
+            dead.instance_variable_set(:@io, nil)
+          end
+          expect { ::IO.select([dead], nil, nil, 0) }.to raise_error(error_class)
+          allow(selector).to receive(:select) do |*args|
+            ::IO.select(*args)
+          end
+
+          io_loop.tick
+
+          expect(dead).to be_closed
+          expect(healthy).not_to be_closed
+          received = nil
+          healthy.on_data {|data| received = data}
+          writer.write('hello')
+          io_loop.tick
+          expect(received).to eq('hello')
+        ensure
+          writer.close if writer
+          begin
+            dead_io.close if dead_io && !dead_io.closed?
+          rescue ::Errno::EBADF
+            nil
+          end
+        end
+      end
+
+      it 'propagates select errors unrelated to dead sockets' do
+        expect(selector).to receive(:select).and_raise(::TypeError, 'bad selector argument')
+        expect { io_loop.tick }.to raise_error(::TypeError, 'bad selector argument')
+      end
+    end
+
+    describe(IoReactor::Connection) do
+      let(:clock) { double('clock', now: 100.0) }
+      let(:socket) { double('socket', close: nil) }
+      let(:socket_impl) do
+        impl = double('socket_impl')
+        allow(impl).to receive(:getaddrinfo).and_return([[nil, 9042, nil, '127.0.0.1', ::Socket::AF_INET, ::Socket::SOCK_STREAM]])
+        allow(impl).to receive(:sockaddr_in).and_return('SOCKADDR')
+        allow(impl).to receive(:new).and_return(socket)
+        impl
+      end
+      let(:connection) do
+        IoReactor::Connection.new('127.0.0.1', 9042, 0.25, double('unblocker', unblock: nil), clock, socket_impl)
+      end
+
+      it 'fails at the connect deadline without waiting for another polling tick' do
+        allow(socket).to receive(:connect_nonblock).and_raise(Errno::EINPROGRESS)
+        future = connection.connect
+        allow(clock).to receive(:now).and_return(100.25)
+
+        connection.connect
+        expect { future.value }.to raise_error(Ione::Io::ConnectionTimeoutError)
+        expect(connection).to be_closed
+      end
+
+      [nil, Errno::EISCONN].each do |result|
+        it "prefers a completed connect (#{result || 'success'}) over the deadline" do
+          allow(socket).to receive(:connect_nonblock).and_raise(Errno::EINPROGRESS)
+          future = connection.connect
+          if result
+            allow(socket).to receive(:connect_nonblock).and_raise(result)
+          else
+            allow(socket).to receive(:connect_nonblock).and_return(0)
+          end
+          allow(clock).to receive(:now).and_return(100.25)
+
+          connection.connect
+
+          expect(future.value).to eq(connection)
+          expect(connection).to be_connected
+        end
+      end
+
+      it 'fails only its own connection when the first connect attempt raises IOError' do
+        allow(socket).to receive(:connect_nonblock).and_raise(IOError, 'closed stream')
+        future = nil
+
+        expect { future = connection.connect }.not_to raise_error
+
+        expect { future.value }.to raise_error(Ione::Io::ConnectionError, 'closed stream')
+        expect(connection).to be_closed
+      end
+    end
+
+    describe(IoReactor::SslConnection) do
+      it 'accepts a handshake that completes at the deadline and preserves the completed future' do
+        clock = double('clock', now: 100.0)
+        raw_socket = double('raw socket', closed?: false, close: nil)
+        ssl_socket = double('SSL socket', close: nil)
+        socket_impl = double('SSL socket implementation', new: ssl_socket)
+        connection = IoReactor::SslConnection.new('127.0.0.1', 9042, raw_socket, double('unblocker', unblock: nil), nil, 100.25, clock)
+        connection.instance_variable_set(:@socket_impl, socket_impl)
+        allow(ssl_socket).to receive(:connect_nonblock).and_raise(::IO::EAGAINWaitReadable)
+        future = connection.connect
+        allow(clock).to receive(:now).and_return(100.25)
+        allow(ssl_socket).to receive(:connect_nonblock).and_return(ssl_socket)
+
+        connection.connect
+
+        expect(future.value).to eq(connection)
+        expect(connection).to be_connected
+        expect(connection.connect).to equal(future)
+        expect(connection).not_to be_closed
+      end
+
+      it 'fails the handshake without raising when the raw socket has disappeared' do
+        connection = IoReactor::SslConnection.new('127.0.0.1', 9042, nil, double('unblocker', unblock: nil), nil,
+                                                  ::Time.now + 1, ::Time)
+        future = connection.connect
+
+        expect(future).to be_failed
+        expect { future.value }.to raise_error(Ione::Io::ConnectionError)
+        expect(connection).to be_closed
+        expect(connection.close).to eq(false)
+      end
+
+      it 'closes the raw socket when closed before the handshake starts' do
+        reader, writer = ::IO.pipe
+        connection = IoReactor::SslConnection.new('127.0.0.1', 9042, reader, double('unblocker', unblock: nil), nil,
+                                                  ::Time.now + 1, ::Time)
+        expect(connection.close).to eq(true)
+        expect(reader).to be_closed
+      ensure
+        reader.close if reader && !reader.closed?
+        writer.close if writer
+      end
+    end
+  end
+end
diff --git a/spec/cassandra/cluster_spec.rb b/spec/cassandra/cluster_spec.rb
index 74b572d01..161f43e4b 100644
--- a/spec/cassandra/cluster_spec.rb
+++ b/spec/cassandra/cluster_spec.rb
@@ -83,10 +83,25 @@ module Cassandra
       end
 
       it 'closes control connection' do
+        expect(io_reactor).not_to receive(:stop)
         expect(control_connection).to receive(:close_async).once.and_return(Ione::Future.resolved)
         expect(cluster.close_async).to eq(promise)
         expect(promise).to have_received(:fulfill).once.with(cluster)
       end
+
+      it 'waits for the control connection to finish stopping the reactor' do
+        stopped = Ione::Promise.new
+        expect(io_reactor).to receive(:stop).once.and_return(stopped.future)
+        allow(executor).to receive(:shutdown)
+
+        cluster.close_async
+        expect(promise).not_to have_received(:fulfill)
+        expect(executor).not_to have_received(:shutdown)
+
+        stopped.fulfill
+        expect(promise).to have_received(:fulfill).once.with(cluster)
+        expect(executor).to have_received(:shutdown).once
+      end
     end
 
     [
@@ -107,5 +122,61 @@ module Cassandra
         end
       end
     end
+
+    # Regression specs for closed and failed clusters releasing their reactor
+    # threads. These use a real reactor and only touch loopback sockets opened
+    # by the spec.
+    context('with a real io reactor') do
+      def free_port
+        server = ::TCPServer.new('127.0.0.1', 0)
+        port   = server.addr[1]
+        server.close
+        port
+      end
+
+      def reactor_threads
+        ::Thread.list.select {|t| t.name == Cluster::IoReactor::THREAD_NAME}
+      end
+
+      describe('#close') do
+        it 'stops the reactor thread and the executor threads' do
+          before  = ::Thread.list.size
+          driver  = Driver.new(io_reactor: Cluster::IoReactor.new)
+          cluster = driver.cluster
+
+          driver.io_reactor.start.value
+          await { reactor_threads.size == 1 }
+          expect(::Thread.list.size).to be > before
+
+          cluster.close
+
+          await { ::Thread.list.size <= before }
+          expect(reactor_threads).to be_empty
+        end
+
+        it 'can be called more than once' do
+          driver  = Driver.new(io_reactor: Cluster::IoReactor.new)
+          cluster = driver.cluster
+          driver.io_reactor.start.value
+
+          expect(cluster.close).to eq(cluster)
+          expect(cluster.close).to eq(cluster)
+          expect(reactor_threads).to be_empty
+        end
+      end
+
+      describe('Cassandra.cluster') do
+        it 'leaves no threads behind when no host can be reached' do
+          before = ::Thread.list.size
+
+          expect do
+            Cassandra.cluster(hosts: ['127.0.0.1'], port: free_port, connect_timeout: 1, timeout: 1)
+          end.to raise_error(Errors::NoHostsAvailable)
+
+          await { ::Thread.list.size <= before }
+          expect(reactor_threads).to be_empty
+        end
+      end
+    end
   end
 end
diff --git a/spec/cassandra/reconnection/policies/exponential_spec.rb b/spec/cassandra/reconnection/policies/exponential_spec.rb
new file mode 100644
index 000000000..661c9d063
--- /dev/null
+++ b/spec/cassandra/reconnection/policies/exponential_spec.rb
@@ -0,0 +1,92 @@
+# encoding: utf-8
+
+#--
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#++
+
+require 'spec_helper'
+
+module Cassandra
+  module Reconnection
+    module Policies
+      describe(Exponential) do
+        describe('#schedule') do
+          it 'grows exponentially up to the ceiling' do
+            schedule = Exponential.new(0.5, 10, 2).schedule
+            expect(Array.new(7) { schedule.next }).to eq([0.5, 1.0, 2.0, 4.0, 8.0, 10, 10])
+          end
+
+          it 'has no jitter by default' do
+            intervals = Array.new(20) { Exponential.new(1, 60, 2).schedule.next }
+            expect(intervals.uniq).to eq([1])
+          end
+
+          context('with jitter') do
+            let(:policy) { Exponential.new(1, 60, 2, jitter: 0.25) }
+
+            it 'randomises each interval within the requested fraction' do
+              first = Array.new(50) { policy.schedule.next }
+              expect(first).to all(be_between(0.75, 1.25))
+              expect(first.uniq.size).to be > 1
+
+              schedule = policy.schedule
+              schedule.next
+              expect(schedule.next).to be_between(1.5, 2.5)
+            end
+
+            it 'never exceeds the ceiling' do
+              schedule  = policy.schedule
+              intervals = Array.new(20) { schedule.next }
+              expect(intervals).to all(be <= 60)
+              expect(intervals.last(5)).to all(be_between(45, 60))
+            end
+
+            it 'uses the given source of randomness' do
+              random = double('random', rand: 1.0)
+              schedule = Exponential.new(1, 60, 2, jitter: 0.25, random: random).schedule
+              expect(schedule.next).to eq(1.25)
+            end
+
+            it 'spreads draws uniformly across the bounded window at the ceiling' do
+              random = double('random')
+              allow(random).to receive(:rand).and_return(0.0, 0.25, 0.5, 0.75)
+              schedule = Exponential.new(60, 60, 2, jitter: 0.25, random: random).schedule
+
+              expect(Array.new(4) { schedule.next }).to eq([45, 48.75, 52.5, 56.25])
+            end
+
+            it 'bounds the window before drawing when approaching the ceiling' do
+              random = double('random', rand: 0.5)
+              schedule = Exponential.new(50, 60, 2, jitter: 0.25, random: random).schedule
+
+              expect(schedule.next).to eq(48.75)
+            end
+
+            it 'rejects a jitter outside 0...1' do
+              expect { Exponential.new(1, 60, 2, jitter: 1) }.to raise_error(::ArgumentError)
+              expect { Exponential.new(1, 60, 2, jitter: -0.1) }.to raise_error(::ArgumentError)
+            end
+
+            it 'rejects invalid jitter values with ArgumentError' do
+              [nil, Object.new, 'invalid', Float::NAN].each do |jitter|
+                expect { Exponential.new(1, 60, 2, jitter: jitter) }.to raise_error(::ArgumentError)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end