diff --git a/lib/resolv.rb b/lib/resolv.rb
index 6b58f92..03e5e9a 100644
--- a/lib/resolv.rb
+++ b/lib/resolv.rb
@@ -562,7 +562,21 @@ def fetch_resource(name, typeclass)
             next if !sender
             senders[[candidate, requester, nameserver, port]] = sender
           end
-          reply, reply_name = requester.request(sender, tout)
+          begin
+            reply, reply_name = requester.request(sender, tout)
+          rescue ResolvTimeout
+            # Giving up part way through a frame loses stream sync, and a peer
+            # seen going away leaves the socket dead.  Either way the requester
+            # says so, and the next attempt has to open a fresh connection.  A
+            # timeout with the stream still on a frame boundary keeps it; a
+            # peer that leaves while nothing is being read goes unnoticed here
+            # and only shows up when the next request is written.
+            unless requester.reusable?
+              requesters.delete([nameserver, port])
+              requester.close
+            end
+            raise
+          end
           case reply.rcode
           when RCode::NoError
             if reply.tc == 1 and not Requester::TCP === requester
@@ -698,6 +712,12 @@ def initialize
         @socks = nil
       end
 
+      # Whether another request may be sent over the same transport.  Only a
+      # stream transport can end up in a state that rules this out.
+      def reusable?
+        true
+      end
+
       def request(sender, tout)
         start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
         timelimit = start + tout
@@ -724,7 +744,7 @@ def request(sender, tout)
             raise ResolvTimeout
           end
           begin
-            reply, from = recv_reply(select_result[0])
+            reply, from = recv_reply(select_result[0], timelimit)
           rescue Errno::ECONNREFUSED, # GNU/Linux, FreeBSD
                  Errno::ECONNRESET, # Windows
                  EOFError
@@ -801,7 +821,7 @@ def lazy_initialize
           self
         end
 
-        def recv_reply(readable_socks)
+        def recv_reply(readable_socks, timelimit = nil)
           lazy_initialize
           reply, from = readable_socks[0].recvfrom(UDPSize)
           return reply, [from[3],from[1]]
@@ -870,7 +890,7 @@ def lazy_initialize
           self
         end
 
-        def recv_reply(readable_socks)
+        def recv_reply(readable_socks, timelimit = nil)
           lazy_initialize
           reply = readable_socks[0].recv(UDPSize)
           return reply, nil
@@ -933,15 +953,28 @@ def initialize(host, port=Port)
           sock = TCPSocket.new(@host, @port)
           @socks = [sock]
           @senders = {}
+          @reusable = true
+        end
+
+        def reusable?
+          @reusable
         end
 
-        def recv_reply(readable_socks)
-          len_data = readable_socks[0].read(2)
+        def recv_reply(readable_socks, timelimit = nil)
+          sock = readable_socks[0]
+          len_data = read_exactly(sock, 2, timelimit)
           raise EOFError if len_data.nil? || len_data.bytesize != 2
           len = len_data.unpack('n')[0]
-          reply = @socks[0].read(len)
+          reply = read_exactly(sock, len, timelimit)
           raise EOFError if reply.nil? || reply.bytesize != len
+          @reusable = true
           return reply, nil
+        rescue EOFError, SystemCallError
+          # Whatever the kernel reported, this socket cannot be trusted for
+          # another frame.  In practice that is the peer closing or resetting;
+          # the rest is rare enough that erring towards reconnecting is right.
+          @reusable = false
+          raise
         end
 
         def sender(msg, data, host=@host, port=@port)
@@ -963,11 +996,40 @@ def send
         end
 
         def close
+          @reusable = false
           super
           @senders.each_key {|from,id|
             DNS.free_request_id(@host, @port, id)
           }
         end
+
+        private
+
+        # Read +len+ bytes, giving up with ResolvTimeout once +timelimit+ (a
+        # CLOCK_MONOTONIC value) has passed.  A shorter result means the peer
+        # closed the connection, which the caller turns into an EOFError.
+        # Consuming any byte marks the requester unusable until the whole frame
+        # has been read, since giving up in between loses frame sync.  Without a
+        # +timelimit+ the read blocks instead and keeps no such mark; that is
+        # only for a caller still using the one argument form of #recv_reply.
+        def read_exactly(sock, len, timelimit)
+          return sock.read(len) unless timelimit
+          buf = String.new
+          while buf.bytesize < len
+            case chunk = sock.read_nonblock(len - buf.bytesize, exception: false)
+            when :wait_readable
+              remaining = timelimit - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+              raise ResolvTimeout if remaining <= 0
+              sock.wait_readable(remaining) or raise ResolvTimeout
+            when nil
+              break
+            else
+              @reusable = false
+              buf << chunk
+            end
+          end
+          buf
+        end
       end
 
       ##
diff --git a/test/resolv/test_dns.rb b/test/resolv/test_dns.rb
index 0b81118..b4ef92b 100644
--- a/test/resolv/test_dns.rb
+++ b/test/resolv/test_dns.rb
@@ -843,6 +843,8 @@ def test_tcp_connection_closed_before_length
           assert_raise(Resolv::ResolvTimeout) do
             requester.request(sender, 2)
           end
+          # The peer went away, so this connection cannot carry another request.
+          assert_equal(false, requester.reusable?)
         ensure
           requester.close
         end
@@ -943,4 +945,314 @@ def test_tcp_connection_closed_with_partial_message_body
       client_thread.join
     end
   end
+
+  def accept_within_timeout(t)
+    Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { t.accept }
+  end
+
+  # Reads one length prefixed DNS message from +sock+.
+  def read_framed_query(sock)
+    len_data = sock.read(2)
+    flunk('the client closed the connection before sending a query') unless len_data&.bytesize == 2
+    Resolv::DNS::Message.decode(sock.read(len_data.unpack('n')[0]))
+  end
+
+  # Builds the encoded reply answering +query+ with a single A record.
+  def reply_for_query(query, address)
+    reply = Resolv::DNS::Message.new(query.id)
+    reply.qr = 1
+    reply.rd = query.rd
+    reply.ra = 1
+    query.each_question do |name, typeclass|
+      reply.add_question(name, typeclass)
+      reply.add_answer(name, 3600, Resolv::DNS::Resource::IN::A.new(address))
+    end
+    reply.encode
+  end
+
+  def framed(encoded)
+    [encoded.bytesize].pack('n') << encoded
+  end
+
+  # Runs a TCP server which replies with +reply+ and then keeps the connection
+  # open until the client side is done, so that only the timeout can end the
+  # request.
+  def with_tcp_server_keeping_connection_open(reply)
+    with_tcp('127.0.0.1', 0) do |t|
+      _, server_port, _, server_address = t.addr
+      done = Thread::Queue.new
+
+      server_thread = Thread.new do
+        ct = accept_within_timeout(t)
+        begin
+          ct.recv(512)
+          ct.write(reply)
+          done.pop
+        ensure
+          ct.close
+        end
+      end
+
+      client_thread = Thread.new do
+        begin
+          yield server_address, server_port
+        ensure
+          done.push(true)
+        end
+      end
+
+      assert_join_threads([client_thread, server_thread])
+    end
+  end
+
+  def request_over_tcp(server_address, server_port, tout)
+    requester = Resolv::DNS::Requester::TCP.new(server_address, server_port)
+    begin
+      msg = Resolv::DNS::Message.new
+      msg.add_question('example.org', Resolv::DNS::Resource::IN::A)
+      sender = requester.sender(msg, msg)
+      Timeout.timeout(EnvUtil.apply_timeout_scale(10)) do
+        requester.request(sender, EnvUtil.apply_timeout_scale(tout))
+      end
+    ensure
+      requester.close
+    end
+  end
+
+  def test_tcp_partial_length_prefix_kept_open
+    with_tcp_server_keeping_connection_open("\x00") do |server_address, server_port|
+      assert_raise(Resolv::ResolvTimeout) do
+        request_over_tcp(server_address, server_port, 0.5)
+      end
+    end
+  end
+
+  def test_tcp_partial_message_body_kept_open
+    reply = [10].pack('n') << '12345' # 5 bytes of a 10 byte message
+    with_tcp_server_keeping_connection_open(reply) do |server_address, server_port|
+      assert_raise(Resolv::ResolvTimeout) do
+        request_over_tcp(server_address, server_port, 0.5)
+      end
+    end
+  end
+
+  def test_tcp_complete_reply_kept_open
+    with_tcp('127.0.0.1', 0) do |t|
+      _, server_port, _, server_address = t.addr
+      done = Thread::Queue.new
+
+      server_thread = Thread.new do
+        ct = accept_within_timeout(t)
+        begin
+          ct.write(framed(reply_for_query(read_framed_query(ct), '192.0.2.1')))
+          done.pop
+        ensure
+          ct.close
+        end
+      end
+
+      client_thread = Thread.new do
+        begin
+          reply, = request_over_tcp(server_address, server_port, 2)
+          assert_equal(1, reply.answer.length)
+          assert_equal('192.0.2.1', reply.answer[0][2].address.to_s)
+        ensure
+          done.push(true)
+        end
+      end
+
+      assert_join_threads([client_thread, server_thread])
+    end
+  end
+
+  def test_tcp_reply_arriving_in_two_chunks
+    with_tcp('127.0.0.1', 0) do |t|
+      _, server_port, _, server_address = t.addr
+      done = Thread::Queue.new
+
+      server_thread = Thread.new do
+        ct = accept_within_timeout(t)
+        begin
+          reply = framed(reply_for_query(read_framed_query(ct), '192.0.2.1'))
+          ct.write(reply.byteslice(0, 6))
+          sleep EnvUtil.apply_timeout_scale(0.2)
+          ct.write(reply.byteslice(6..-1))
+          done.pop
+        ensure
+          ct.close
+        end
+      end
+
+      client_thread = Thread.new do
+        begin
+          reply, = request_over_tcp(server_address, server_port, 2)
+          assert_equal(1, reply.answer.length)
+          assert_equal('192.0.2.1', reply.answer[0][2].address.to_s)
+        ensure
+          done.push(true)
+        end
+      end
+
+      assert_join_threads([client_thread, server_thread])
+    end
+  end
+
+  def test_truncated_tcp_fallback_with_partial_message_body_kept_open
+    with_udp_and_tcp('127.0.0.1', 0) do |u, t|
+      _, server_port, _, server_address = u.addr
+      done = Thread::Queue.new
+
+      client_thread = Thread.new do
+        begin
+          dns = Resolv::DNS.new(nameserver_port: [[server_address, server_port]],
+                                raise_timeout_errors: true)
+          begin
+            dns.timeouts = EnvUtil.apply_timeout_scale(0.5)
+            assert_raise(Resolv::ResolvError) do
+              Timeout.timeout(EnvUtil.apply_timeout_scale(10)) do
+                dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A)
+              end
+            end
+          ensure
+            dns.close
+          end
+        ensure
+          done.push(true)
+        end
+      end
+
+      udp_server_thread = Thread.new do
+        msg, (_, client_port, _, client_address) =
+          Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) }
+        id, word2, = msg.unpack('nnnnnn')
+        opcode = (word2 & 0x7800) >> 11
+        rd = (word2 & 0x0100) >> 8
+        qr = 1
+        tc = 1 # ask the client to retry over TCP
+        ra = 1
+        word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7)
+        u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port)
+      end
+
+      tcp_server_thread = Thread.new do
+        ct = accept_within_timeout(t)
+        begin
+          ct.recv(512)
+          ct.write([10].pack('n') << '12345') # 5 bytes of a 10 byte message
+          done.pop
+        ensure
+          ct.close
+        end
+      end
+
+      assert_join_threads([client_thread, udp_server_thread, tcp_server_thread])
+    end
+  end
+
+  # A frame read that gives up part way through a frame leaves the stream
+  # between frame boundaries, so the retry has to start from a new connection.
+  def test_truncated_tcp_fallback_retries_on_a_new_connection
+    with_udp_and_tcp('127.0.0.1', 0) do |u, t|
+      _, server_port, _, server_address = u.addr
+
+      client_thread = Thread.new do
+        Resolv::DNS.open(nameserver_port: [[server_address, server_port]],
+                         raise_timeout_errors: true) do |dns|
+          dns.timeouts = [EnvUtil.apply_timeout_scale(1),
+                          EnvUtil.apply_timeout_scale(3)]
+          Timeout.timeout(EnvUtil.apply_timeout_scale(20)) do
+            dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A)
+          end
+        end
+      end
+
+      udp_server_thread = Thread.new do
+        msg, (_, client_port, _, client_address) =
+          Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) }
+        id, word2, = msg.unpack('nnnnnn')
+        opcode = (word2 & 0x7800) >> 11
+        rd = (word2 & 0x0100) >> 8
+        qr = 1
+        tc = 1 # ask the client to retry over TCP
+        ra = 1
+        word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7)
+        u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port)
+      end
+
+      tcp_server_thread = Thread.new do
+        partial = accept_within_timeout(t)
+        begin
+          read_framed_query(partial)
+          partial.write([45].pack('n') << 'abcdef') # 6 bytes of a 45 byte message
+          complete = accept_within_timeout(t)
+          begin
+            complete.write(framed(reply_for_query(read_framed_query(complete), '192.0.2.1')))
+          ensure
+            complete.close
+          end
+        ensure
+          partial.close
+        end
+      end
+
+      result, = assert_join_threads([client_thread, udp_server_thread, tcp_server_thread])
+      assert_equal(['192.0.2.1'], result.map {|rr| rr.address.to_s })
+    end
+  end
+
+  # A timeout with no bytes read leaves the stream on a frame boundary, so the
+  # retry keeps the connection.  The reply below only ever reaches the client if
+  # the second attempt reuses the socket the first one opened.
+  def test_truncated_tcp_fallback_keeps_the_connection_when_nothing_arrived
+    with_udp_and_tcp('127.0.0.1', 0) do |u, t|
+      _, server_port, _, server_address = u.addr
+      done = Thread::Queue.new
+
+      client_thread = Thread.new do
+        begin
+          Resolv::DNS.open(nameserver_port: [[server_address, server_port]],
+                           raise_timeout_errors: true) do |dns|
+            dns.timeouts = [EnvUtil.apply_timeout_scale(0.5),
+                            EnvUtil.apply_timeout_scale(5)]
+            Timeout.timeout(EnvUtil.apply_timeout_scale(20)) do
+              dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A)
+            end
+          end
+        ensure
+          done.push(true)
+        end
+      end
+
+      udp_server_thread = Thread.new do
+        msg, (_, client_port, _, client_address) =
+          Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) }
+        id, word2, = msg.unpack('nnnnnn')
+        opcode = (word2 & 0x7800) >> 11
+        rd = (word2 & 0x0100) >> 8
+        qr = 1
+        tc = 1 # ask the client to retry over TCP
+        ra = 1
+        word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7)
+        u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port)
+      end
+
+      tcp_server_thread = Thread.new do
+        ct = accept_within_timeout(t)
+        begin
+          query = read_framed_query(ct)
+          # Stay silent past the first interval, then answer on this connection.
+          # It has to stay open until the client is done: the retry resends the
+          # query, and closing with that still unread would discard the reply.
+          sleep EnvUtil.apply_timeout_scale(1)
+          ct.write(framed(reply_for_query(query, '192.0.2.1')))
+          done.pop
+        ensure
+          ct.close
+        end
+      end
+
+      result, = assert_join_threads([client_thread, udp_server_thread, tcp_server_thread])
+      assert_equal(['192.0.2.1'], result.map {|rr| rr.address.to_s })
+    end
+  end
 end