diff --git a/DOCKER.md b/DOCKER.md
index f124a28..85a1ed6 100644
--- a/DOCKER.md
+++ b/DOCKER.md
@@ -100,7 +100,18 @@ docker exec nginx-proxy nginx -s reload
 This configuration:
 - Enables HTTP/1.1 for the upstream connection (required for WebSocket upgrade)
 - Passes the `Upgrade` and `Connection` headers to the backend
-- Sets a long read timeout for persistent WebSocket connections
+- Sets a long read timeout (nginx's default is 60s) so downloads waiting on a
+  slow sender and idle WebSockets aren't cut
+
+**Notes:**
+- The application accepts any casing of the `Connection` header (nginx-proxy
+  sends `upgrade` in lowercase), so `proxy_set_header Connection "Upgrade"` is
+  only kept for older application versions.
+- Downloads are sent with `X-Accel-Buffering: no`, which makes nginx stream
+  them straight through instead of buffering up to 1GB to disk
+  (`proxy_max_temp_file_size`). Backpressure then flows all the way from the
+  downloader to the sender's browser, so the server never holds more than a
+  few MB of a transfer in memory.
 
 ## Volumes
 
@@ -158,6 +169,13 @@ Ensure:
    docker exec nginx-proxy nginx -s reload
    ```
 
+### Clients get `429 Too Many Connections` on `/ws`
+
+The application limits concurrent WebSocket connections per client IP
+(`DLCENTER_MAX_CONNECTIONS_PER_IP`, default 50). Behind a NAT, a whole office
+shares one IP. Check the application logs for `WebSocket handshake failed`
+messages; restarting the container resets the counters.
+
 ### Container can't connect to nginx-proxy
 
 Ensure all containers are on the same network:
diff --git a/app.rb b/app.rb
index 6bb61fa..a7ce532 100644
--- a/app.rb
+++ b/app.rb
@@ -4,6 +4,7 @@
 require 'securerandom'
 require 'json'
 require 'dlcenter'
+require 'dlcenter/thin_ipv6_host'
 
 module DLCenter
   class App < Sinatra::Base
@@ -30,7 +31,6 @@ class App < Sinatra::Base
       end
     end
 
-    # CSRF protection helper
     helpers do
       def sanitize_filename(filename)
         return 'download' if filename.nil? || filename.empty?
@@ -93,37 +93,86 @@ def check_csrf
 
         halt 403, 'CSRF check failed'
       end
+
+      def share_headers(share)
+        options = {
+          "Cache-Control" => "no-cache, private",
+          "Pragma"        => "no-cache",
+          "Content-type"  => safe_content_type(share.content_type),
+          "Content-Disposition" => "attachment; filename=\"#{sanitize_filename(share.name)}\""
+        }
+        options["Content-Length"] = "#{share.size}" unless share.size.nil?
+        headers options
+      end
+
+      # Streams a download body fed by a sender, with backpressure.
+      # The block receives (out, connection, closer):
+      # - connection: the downloader's EventMachine connection (nil outside Thin),
+      #   used to pace the sender on what the downloader actually consumed
+      # - closer: deferrable fired when the downloader disconnects
+      def stream_download
+        # Tell nginx to pass data straight through instead of buffering the
+        # whole response to disk (which defeats backpressure).
+        headers 'X-Accel-Buffering' => 'no'
+        connection = env['async.connection']
+        closer = env['async.close']
+        # Thin drops connections idle for 30s. A download waiting on a slow or
+        # paused sender is legitimately idle; sender liveness is handled by
+        # the WebSocket heartbeat.
+        if connection.respond_to?(:comm_inactivity_timeout=)
+          connection.comm_inactivity_timeout = 0
+        end
+        stream(:keep_open) do |out|
+          yield(out, connection, closer)
+          nil
+        end
+      end
     end
 
     get '/ws' do
-      check_rate_limit(request.ip)
+      # em-websocket 0.3.x only accepts "Connection: Upgrade" with that exact
+      # casing. Proxies (nginx-proxy sends "upgrade") and some clients differ,
+      # so normalize what we hand over to it.
+      request.env['HTTP_CONNECTION'] = 'Upgrade' if request.websocket?
+
+      ip = request.ip
+      check_rate_limit(ip)
+      namespace = namespace_for_request(request)
+      client = nil
       begin
-          request.websocket do |ws|
-            namespace = namespace_for_request(request)
-            client = WSClient.new namespace, ws
-            namespace.add_client client
-            ws.onclose do
-              release_rate_limit(request.ip)
-            end
-          end
+        request.websocket do |ws|
+          client = WSClient.new(namespace, ws) { release_rate_limit(ip) }
+          namespace.add_client client
+        end
       rescue SinatraWebsocket::Error::ConnectionError
-          release_rate_limit(request.ip)
-          puts "Not a websocket"
+        release_rate_limit(ip)
+        halt 400, 'Not a websocket request'
+      rescue StandardError => e
+        # Handshake failed (e.g. EventMachine::WebSocket::HandshakeError):
+        # the onclose callback will never fire, so clean up here.
+        namespace.remove_client(client) if client
+        release_rate_limit(ip)
+        puts "WebSocket handshake failed: #{e.class}: #{e.message}"
+        halt 400, 'WebSocket handshake failed'
       end
     end
 
     post '/p/:filename' do
       check_csrf
-      check_rate_limit(request.ip)
+      ip = request.ip
+      check_rate_limit(ip)
       stream(:keep_open) do |out|
-        namespace = namespace_for_request(request)
-        client = IOClient.new namespace, request.env['data.input'], out,
-          filename: sanitize_filename(params[:filename]),
-          size: request.env["CONTENT_LENGTH"],
-          content_type: safe_content_type(request.env["CONTENT_TYPE"])
-        namespace.add_client client
-        namespace.broadcast_available_shares
-        release_rate_limit(request.ip)
+        begin
+          namespace = namespace_for_request(request)
+          client = IOClient.new namespace, request.env['data.input'], out,
+            filename: sanitize_filename(params[:filename]),
+            size: request.env["CONTENT_LENGTH"],
+            content_type: safe_content_type(request.env["CONTENT_TYPE"])
+          namespace.add_client client
+          namespace.broadcast_available_shares
+        ensure
+          release_rate_limit(ip)
+        end
       end
     end
 
@@ -139,61 +188,36 @@ def namespace_for_request(request)
     get '/g' do
       namespace = namespace_for_request(request)
       share = namespace.shares.first
-      if share
-        options = {
-          "Cache-Control" => "no-cache, private",
-          "Pragma"        => "no-cache",
-          "Content-type"  => safe_content_type(share.content_type),
-          "Content-Disposition" => "attachment; filename=\"#{sanitize_filename(share.name)}\""
-        }
-        options["Content-Length"] = "#{share.size}" unless share.size.nil?
-        headers options
-        stream(:keep_open) do |out|
-          share.content(out)
-          nil
-        end
-      else
-        status 404
+      halt 404 unless share
+      share_headers(share)
+      stream_download do |out, connection, closer|
+        share.content(out, connection: connection, closer: closer)
       end
     end
+
     get '/all' do
       namespace = namespace_for_request(request)
       shares = namespace.shares
-      if shares.size > 0
-        options = {
-          "Cache-Control" => "no-cache, private",
-          "Pragma"        => "no-cache",
-          "Content-type"  => "application/zip",
-          "Content-Disposition" => "attachment; filename=\"dlcenter-pack-#{Time.now.strftime '%F'}.zip\""
-        }
-        headers options
-        stream(:keep_open) do |out|
-          Share.content(shares, out)
-          nil
-        end
-      else
-        status 404
+      halt 404 if shares.empty?
+      headers \
+        "Cache-Control" => "no-cache, private",
+        "Pragma"        => "no-cache",
+        "Content-type"  => "application/zip",
+        "Content-Disposition" => "attachment; filename=\"dlcenter-pack-#{Time.now.strftime '%F'}.zip\""
+      stream_download do |out, connection, closer|
+        Share.content(shares, out, connection: connection, closer: closer)
       end
     end
+
     get '/share/:uuid' do
       uuid = params[:uuid]
       # Validate UUID format to prevent log injection
       halt 400, 'Invalid UUID' unless uuid.match?(/\A[a-f0-9\-]{36}\z/i)
       share = settings.registry.get_share_by_uuid uuid
-      if share
-        headers \
-        "Cache-Control" => "no-cache, private",
-        "Pragma"        => "no-cache",
-        "Content-type"  => safe_content_type(share.content_type),
-        "Content-Length" => "#{share.size}",
-        "Content-Disposition" => "attachment; filename=\"#{sanitize_filename(share.name)}\""
-
-        stream(:keep_open) do |out|
-          share.content(out)
-          nil
-        end
-      else
-        status 404
+      halt 404 unless share
+      share_headers(share)
+      stream_download do |out, connection, closer|
+        share.content(out, connection: connection, closer: closer)
       end
     end
   end
diff --git a/lib/dlcenter/client.rb b/lib/dlcenter/client.rb
index 423b444..ccfd6ee 100644
--- a/lib/dlcenter/client.rb
+++ b/lib/dlcenter/client.rb
@@ -42,6 +42,22 @@ def ask_for_stream(stream)
       send_msg(:stream, uuid: stream.uuid, share: stream.share.uuid)
       return stream
     end
+    def active_streams
+      @streams
+    end
+    # The downloader of `stream` went away: forget the stream and tell the
+    # sender to stop pushing data for it.
+    def abort_stream(stream)
+      return unless @streams.delete(stream.uuid)
+      stream.abort
+      send_msg(:stream_close, uuid: stream.uuid)
+    end
+    # The sender is gone: end every download it was feeding.
+    def close_streams
+      streams = @streams.values
+      @streams.clear
+      streams.each(&:close)
+    end
     def send_msg(msg, options={})
       raise NotImplementedError.new(msg)
     end
@@ -60,17 +76,24 @@ def initialize namespace, io_in, io_out, options = {}
           })
         self.add_share(share)
     end
+    # Runs in a worker thread (Sinatra streams are deferred on EventMachine):
+    # pace the reads on the downloader actually consuming the data.
     def flush_io(uuid)
       stream = @streams[uuid]
+      drained = Queue.new
       begin
         while (data = @io_in.read 1024*1024)
+          break if stream.closed?
           stream.got_chunk(data)
           stream.drain_buffer
+          stream.when_drained { drained.push(true) }
+          drained.pop
         end
       rescue IOError => e
         puts "Error while flushing #{e}"
       end
       stream.close
+      @streams.delete(uuid)
       @io_out.close
       @namespace.remove_client(self)
     end
@@ -79,6 +102,7 @@ def send_msg(msg, params={})
       when :shares then nil
       when :hello  then nil
       when :stream then flush_io(params[:uuid])
+      when :stream_close then nil
       else
         raise "Invalid msg type #{msg} with params #{params}"
       end
@@ -87,51 +111,76 @@ def send_msg(msg, params={})
 
   class WSClient < Client
     HEARTBEAT_INTERVAL = 30  # seconds between pings
-    HEARTBEAT_TIMEOUT = 10   # seconds to wait for pong
+    HEARTBEAT_TIMEOUT = 30   # seconds to wait for pong (slow uplinks may delay it behind chunks)
 
-    def initialize namespace, ws
+    # `on_close` is invoked once when the socket is gone (after the client
+    # has been removed from its namespace).
+    def initialize namespace, ws, &on_close
       super(namespace)
       @ws = ws
-      @pong_received = true
+      @on_close = on_close
+      @alive = true
       @heartbeat_timer = nil
       @timeout_timer = nil
 
       ws.onopen do
-        self.send_msg(:hello, text: "Hello World!")
-        self.send_msg(:shares, shares: @namespace.get_shares_json)
-        start_heartbeat
+        guard("onopen") do
+          self.send_msg(:hello, text: "Hello World!")
+          self.send_msg(:shares, shares: @namespace.get_shares_json)
+          start_heartbeat
+        end
       end
       ws.onmessage do |tmsg|
-
-        begin
-          msg = JSON.parse(tmsg, symbolize_names: true)
-        rescue
-          puts "Can't parse JSON message"
+        guard("onmessage") do
+          # Any traffic proves the client is alive
+          @alive = true
+          begin
+            msg = JSON.parse(tmsg, symbolize_names: true)
+          rescue JSON::ParserError
+            puts "Can't parse JSON message"
+            next
+          end
+          self.handle_ws_msg(msg) if msg.is_a?(Hash)
         end
-        self.handle_ws_msg(msg)
-
       end
       ws.onclose do
-        puts "WS closed"
-        stop_heartbeat
-        @namespace.remove_client(self)
+        guard("onclose") do
+          puts "WS closed"
+          stop_heartbeat
+          @namespace.remove_client(self)
+          @on_close.call if @on_close
+        end
       end
     end
 
+    # A raised exception in an EventMachine callback or timer takes the whole
+    # server down with it (and every connected user). Never let that happen
+    # because of one misbehaving client.
+    def guard(context)
+      yield
+    rescue StandardError => e
+      puts "Error in WebSocket #{context}: #{e.class}: #{e.message}"
+      puts e.backtrace.first(5).join("\n") if e.backtrace
+    end
+
     def start_heartbeat
-      @heartbeat_timer = EM.add_periodic_timer(HEARTBEAT_INTERVAL) do
-        if @pong_received
-          @pong_received = false
-          send({type: :ping}.to_json)
-          @timeout_timer = EM.add_timer(HEARTBEAT_TIMEOUT) do
-            unless @pong_received
-              puts "Heartbeat timeout, closing connection"
-              @ws.close
+      @heartbeat_timer = EM.add_periodic_timer(self.class::HEARTBEAT_INTERVAL) do
+        guard("heartbeat") do
+          if @alive
+            @alive = false
+            send({type: :ping}.to_json)
+            @timeout_timer = EM.add_timer(self.class::HEARTBEAT_TIMEOUT) do
+              guard("heartbeat timeout") do
+                unless @alive
+                  puts "Heartbeat timeout, closing connection"
+                  close_ws
+                end
+              end
             end
+          else
+            puts "No pong received, closing connection"
+            close_ws
           end
-        else
-          puts "No pong received, closing connection"
-          @ws.close
         end
       end
     end
@@ -143,11 +192,23 @@ def stop_heartbeat
       @timeout_timer = nil
     end
 
+    def close_ws
+      stop_heartbeat
+      if @ws.respond_to?(:close_websocket)
+        @ws.close_websocket
+      elsif @ws.respond_to?(:close)
+        @ws.close
+      end
+    rescue StandardError => e
+      puts "Can't close websocket: #{e.message}"
+    end
+
     def send_msg(msg, params={})
       case msg
       when :shares then send({type: :shares, shares: params[:shares]}.to_json)
       when :hello  then send({type: :hello, text: params[:text]}.to_json)
       when :stream then send({type: :stream}.merge(params).to_json)
+      when :stream_close then send({type: :stream_close, uuid: params[:uuid]}.to_json)
       else
         raise "Invalid msg type #{msg} with params #{params}"
       end
@@ -156,7 +217,7 @@ def send_msg(msg, params={})
     def send(ws_msg)
       begin
         @ws.send(ws_msg)
-      rescue
+      rescue StandardError
         puts "Can't send message to #{@ws}"
       end
     end
@@ -249,27 +310,38 @@ def handle_chunk(msg)
       end
 
       stream = @streams[uuid]
-      if stream
-        chunk = Base64.decode64(encoded_chunk)
-        # Validate chunk size
-        if chunk.length > MAX_CHUNK_SIZE
-          puts "Chunk too large: #{chunk.length} bytes"
-          return false
-        end
+      unless stream
+        puts "Unknown stream #{uuid}"
+        return false
+      end
+
+      chunk = Base64.decode64(encoded_chunk)
+      # Validate chunk size
+      if chunk.length > MAX_CHUNK_SIZE
+        puts "Chunk too large: #{chunk.length} bytes"
+        return false
+      end
+
+      begin
         stream.got_chunk(chunk)
-        begin
-          stream.drain_buffer
-          if msg[:close] then
-            stream.close
-          end
-          return true
-        rescue IOError
-          puts "ERROR: can't send data to client"
-        end
+        stream.drain_buffer
+      rescue IOError
+        puts "ERROR: can't send data to client"
+        @streams.delete(uuid)
+        return false
+      end
+
+      if msg[:close]
+        stream.close
+        @streams.delete(uuid)
       else
-        puts "Unknown stream #{uuid}"
+        # Flow control: only ask the sender for more once the downloader has
+        # actually consumed what we already have.
+        stream.when_drained do
+          send({type: :ack, uuid: uuid}.to_json) unless stream.closed?
+        end
       end
-      return false
+      return true
     end
 
     def handle_ws_msg(msg)
@@ -279,7 +351,7 @@ def handle_ws_msg(msg)
       when 'unregister_share' then handle_unregister_share(msg)
       when 'chunk' then handle_chunk(msg)
       when 'ping' then send({type: :pong}.to_json)
-      when 'pong' then @pong_received = true
+      when 'pong' then nil # liveness already recorded in onmessage
       else puts "Unkown msg : #{msg}"
       end
     end
diff --git a/lib/dlcenter/registry.rb b/lib/dlcenter/registry.rb
index c535d6f..3d823e8 100644
--- a/lib/dlcenter/registry.rb
+++ b/lib/dlcenter/registry.rb
@@ -11,6 +11,9 @@ def add_client(client)
     end
 
     def remove_client(client)
+      # Downloads in progress from this client can't complete anymore:
+      # close them so downloaders don't hang waiting for data.
+      client.close_streams if client.respond_to?(:close_streams)
       @clients.delete client
       broadcast_available_shares
     end
@@ -76,13 +79,6 @@ def each_namespace
         end
       end
     end
-    # def each_share
-    #   each_namespace do |namespace|
-    #     namespace.shares.each do |share|
-    #       yield share
-    #     end
-    #   end
-    # end
     def share_count
       count = 0
       each_namespace do |namespace|
@@ -90,9 +86,6 @@ def share_count
       end
       count
     end
-    # def to_s
-    #   "Registry (#{share_count} shares)"
-    # end
     def get_share_by_uuid(uuid)
       # TODO: create a weak cache for retrieving in O(1) the share
       each_namespace do |namespace|
diff --git a/lib/dlcenter/share.rb b/lib/dlcenter/share.rb
index 0aa5f22..0b58043 100644
--- a/lib/dlcenter/share.rb
+++ b/lib/dlcenter/share.rb
@@ -19,7 +19,7 @@ def initialize client, options = {}
       raise "Must have a name" unless self.name
     end
 
-    def self.content(shares, out)
+    def self.content(shares, out, connection: nil, closer: nil)
       w = ZipTricks::BlockWrite.new { |chunk| out.write(chunk) }
       ZipTricks::Streamer.open(w) do |zip|
         shares.each do |share|
@@ -27,7 +27,7 @@ def self.content(shares, out)
           safe_name = sanitize_zip_filename(share.name)
           zip.write_deflated_file(safe_name) do |sink|
             r, w = IO.pipe
-            share.content(w)
+            share.content(w, connection: connection, closer: closer)
             while true
               buffer = r.read(65536)
               sink << buffer
@@ -49,8 +49,14 @@ def self.sanitize_zip_filename(name)
       safe.empty? ? 'file' : safe
     end
 
-    def content(out)
-      Streamer.new(self, out).tap do |stream|
+    # Starts streaming this share to `out`. `connection` (the downloader's
+    # EventMachine connection) enables backpressure; `closer` (deferrable
+    # fired when the downloader disconnects) lets us stop the sender early.
+    def content(out, connection: nil, closer: nil)
+      Streamer.new(self, out, connection: connection).tap do |stream|
+        if closer.respond_to?(:callback)
+          closer.callback { client.abort_stream(stream) }
+        end
         client.ask_for_stream(stream)
       end
     end
diff --git a/lib/dlcenter/streamer.rb b/lib/dlcenter/streamer.rb
index 5401340..f13d3a6 100644
--- a/lib/dlcenter/streamer.rb
+++ b/lib/dlcenter/streamer.rb
@@ -1,17 +1,33 @@
 module DLCenter
+  # A Streamer pipes chunks received from a sender (WebSocket or IO client)
+  # to a downloader's HTTP response body (`out`).
+  #
+  # Flow control: when constructed with the EventMachine `connection` of the
+  # downloader, `when_drained` only fires once the connection's outbound
+  # buffer is below HIGH_WATER. Senders use it to pace themselves so the
+  # server never buffers more than a small window of a large file in memory.
   class Streamer
     # Max buffer size: 10MB - prevents memory exhaustion
     MAX_BUFFER_SIZE = 10 * 1024 * 1024
+    # Outbound bytes queued on the downloader socket above which senders are paused
+    HIGH_WATER = 4 * 1024 * 1024
+    # How often to re-check the outbound buffer while paused (seconds)
+    POLL_INTERVAL = 0.05
 
     attr_reader :share, :buffer, :uuid, :out
-    def initialize(share, out)
+    def initialize(share, out, connection: nil)
       @uuid = SecureRandom.uuid
       @share = share
       @out = out
+      @connection = connection
       @buffer = ""
       @closed = false
     end
 
+    def closed?
+      @closed
+    end
+
     def got_chunk(chunk)
       return if @closed
       # Check buffer size limit
@@ -36,6 +52,26 @@ def drain_buffer
       }
     end
 
+    # Bytes handed to the downloader socket but not yet written to the network.
+    def pending_bytes
+      return 0 unless @connection.respond_to?(:get_outbound_data_size)
+      @connection.get_outbound_data_size
+    rescue StandardError
+      0
+    end
+
+    def drained?
+      pending_bytes < HIGH_WATER
+    end
+
+    # Yields once everything scheduled so far has been flushed far enough
+    # down the downloader socket (or once the stream is closed, so callers
+    # waiting on it never hang). Safe to call from any thread.
+    def when_drained(&blk)
+      EM.next_tick { wait_drained(&blk) }
+    end
+
+    # Normal end of stream: flush and close the downloader response.
     def close
       return if @closed
       @closed = true
@@ -43,5 +79,21 @@ def close
         @out.close
       }
     end
+
+    # Downloader went away: stop accepting data, nothing more to write.
+    def abort
+      @closed = true
+      @buffer = ""
+    end
+
+    private
+
+    def wait_drained(&blk)
+      if @closed || drained?
+        blk.call
+      else
+        EM.add_timer(POLL_INTERVAL) { wait_drained(&blk) }
+      end
+    end
   end
 end
diff --git a/lib/dlcenter/thin_ipv6_host.rb b/lib/dlcenter/thin_ipv6_host.rb
new file mode 100644
index 0000000..a20c440
--- /dev/null
+++ b/lib/dlcenter/thin_ipv6_host.rb
@@ -0,0 +1,23 @@
+require 'thin'
+
+module DLCenter
+  # Thin's C parser splits the Host header on the first ':' to fill
+  # SERVER_NAME / SERVER_PORT, which mangles bracketed IPv6 hosts
+  # ("[::1]:8080" becomes name "[" and port ":1]:8080"). Rack::Lint (enabled
+  # by rackup in development) then rejects the request with a 500.
+  module ThinIPv6Host
+    IPV6_HOST = /\A(\[[^\]]+\])(?::(\d+))?\z/
+
+    def parse(data)
+      result = super
+      host = @env['HTTP_HOST']
+      if host && host.start_with?('[') && (m = IPV6_HOST.match(host))
+        @env['SERVER_NAME'] = m[1]
+        @env['SERVER_PORT'] = m[2] || (@env['HTTPS'] == 'on' ? '443' : '80')
+      end
+      result
+    end
+  end
+end
+
+Thin::Request.prepend(DLCenter::ThinIPv6Host)
diff --git a/public/index.html b/public/index.html
index a493014..afca362 100644
--- a/public/index.html
+++ b/public/index.html
@@ -30,15 +30,15 @@
   <script src="js/libs/filesize.js"></script>
 
   <!-- App utilities -->
-  <script src="js/utils.js"></script>
+  <script src="js/utils.js?v=2"></script>
 </head>
 
 <body class='dropzone'>
   <div id="root" class='main-view'></div>
 
   <!-- React components (transpiled by Babel in browser) -->
-  <script type="text/babel" src="js/components.js"></script>
-  <script type="text/babel" src="js/app.js"></script>
+  <script type="text/babel" src="js/components.js?v=2"></script>
+  <script type="text/babel" src="js/app.js?v=2"></script>
 </body>
 
 </html>
diff --git a/public/js/app.js b/public/js/app.js
index d3b6928..8b3c78c 100644
--- a/public/js/app.js
+++ b/public/js/app.js
@@ -8,6 +8,9 @@ function App() {
   const [textValue, setTextValue] = useState('');
   const wsRef = useRef(null);
   const pingRef = useRef(null);
+  const reconnectTimerRef = useRef(null);
+  const reconnectDelayRef = useRef(1000);
+  const setupRef = useRef(null);
   // Keep a synchronous ref of local shares for streaming lookups
   const localSharesRef = useRef({});
   const downloadHost = `${document.location.protocol}//${document.location.host}`;
@@ -101,16 +104,70 @@ function App() {
     }
   }, []);
 
+  // Announce every local share to the server (on connect and after every
+  // reconnect, since the server forgets us when the socket drops).
+  const registerLocalShares = useCallback((ws) => {
+    Object.values(localSharesRef.current).forEach((share) => {
+      if (share.file) {
+        ws.send(JSON.stringify({
+          type: "register_share",
+          uuid: share.uuid,
+          name: share.name,
+          content_type: share.type,
+          size: share.size
+        }));
+      } else if (share.content) {
+        ws.send(JSON.stringify({
+          type: "register_share",
+          uuid: share.uuid,
+          name: ellipseAt(share.content, 100),
+          content: share.content,
+          content_type: "text/plain",
+          size: share.size
+        }));
+      }
+    });
+  }, []);
+
+  // Exactly one reconnect attempt is ever pending, with exponential backoff.
+  // (Scheduling one from both onerror and onclose doubled the number of
+  // sockets on every failure and saturated the server's per-IP limit.)
+  const scheduleReconnect = useCallback(() => {
+    if (reconnectTimerRef.current) return;
+    const delay = reconnectDelayRef.current;
+    reconnectDelayRef.current = Math.min(delay * 2, 30000);
+    console.log("Reconnecting in " + delay + "ms");
+    reconnectTimerRef.current = setTimeout(() => {
+      reconnectTimerRef.current = null;
+      if (setupRef.current) setupRef.current();
+    }, delay);
+  }, []);
+
   const setupWebSocket = useCallback(() => {
+    const previous = wsRef.current;
+    if (previous && previous.readyState !== WebSocket.CLOSED) {
+      previous.onclose = null;
+      previous.onerror = null;
+      previous.close();
+    }
+    if (pingRef.current) {
+      clearInterval(pingRef.current);
+      pingRef.current = null;
+    }
+
     const protocol = document.location.protocol === "https:" ? "wss:" : "ws:";
     const ws = new WebSocket(protocol + '//' + window.location.host + "/ws");
     wsRef.current = ws;
 
     ws.onopen = () => {
       setConnected(true);
+      reconnectDelayRef.current = 1000;
       console.log('websocket opened');
+      registerLocalShares(ws);
       pingRef.current = setInterval(() => {
-        ws.send(JSON.stringify({ type: "ping" }));
+        if (ws.readyState === WebSocket.OPEN) {
+          ws.send(JSON.stringify({ type: "ping" }));
+        }
       }, 10000);
     };
 
@@ -118,21 +175,20 @@ function App() {
       console.log("Websocket closed");
       setConnected(false);
       setRemoteShares([]);
+      streamAbortAll();
       if (pingRef.current) {
         clearInterval(pingRef.current);
+        pingRef.current = null;
       }
-      setTimeout(setupWebSocket, 2000);
+      scheduleReconnect();
     };
 
     ws.onerror = () => {
+      // A close event always follows an error; reconnection is handled there.
       console.log("Websocket error");
-      setConnected(false);
-      setRemoteShares([]);
-      setTimeout(setupWebSocket, 10000);
     };
 
     ws.onmessage = (m) => {
-      console.log('websocket message: ' + m.data);
       const msg = JSON.parse(m.data);
       switch (msg.type) {
         case "shares":
@@ -144,28 +200,41 @@ function App() {
         case "stream":
           handleStream(msg);
           break;
+        case "ack":
+          streamAck(msg.uuid);
+          break;
+        case "stream_close":
+          streamAbort(msg.uuid);
+          break;
         case "ping":
           ws.send(JSON.stringify({ type: "pong" }));
           break;
         case "pong":
           break;
         default:
-          console.error("Unknown message: " + msg.type);
+          console.warn("Unknown message: " + msg.type);
       }
     };
-  }, [handleStream]);
+  }, [handleStream, registerLocalShares, scheduleReconnect]);
+  setupRef.current = setupWebSocket;
 
   useEffect(() => {
     setupWebSocket();
     return () => {
+      if (reconnectTimerRef.current) {
+        clearTimeout(reconnectTimerRef.current);
+        reconnectTimerRef.current = null;
+      }
       if (wsRef.current) {
+        wsRef.current.onclose = null;
         wsRef.current.close();
       }
       if (pingRef.current) {
         clearInterval(pingRef.current);
+        pingRef.current = null;
       }
     };
-  }, [setupWebSocket]);
+  }, []);
 
   useEffect(() => {
     const dropZone = document.querySelector('.dropzone');
diff --git a/public/js/utils.js b/public/js/utils.js
index b594f7d..70d01c2 100644
--- a/public/js/utils.js
+++ b/public/js/utils.js
@@ -11,24 +11,74 @@ function ellipseAt(str, length) {
   return str.length > length ? str.substring(0, length) + "..." : str;
 }
 
-function streamChunk(share, streamUuid, start, length, ws, cb) {
+// --- Upload streaming with flow control -------------------------------------
+//
+// A share is streamed to the server in chunks over the WebSocket. The server
+// acknowledges each chunk once the downloader has actually consumed it, and we
+// keep at most STREAM_WINDOW chunks unacknowledged. Without this, a big file
+// would be read entirely into the socket's send queue (gigabytes of memory in
+// the tab, the same on the server) and the heartbeat pong would be stuck
+// behind it, getting the connection killed.
+
+const STREAM_CHUNK_SIZE = 1024000;
+const STREAM_WINDOW = 4;                    // chunks in flight before waiting for acks
+const STREAM_MAX_BUFFERED = 8 * 1024 * 1024; // don't push more if the socket queue is this big
+const activeStreams = {};
+
+function readChunkAsBase64(file, start, length, cb) {
   const reader = new FileReader();
-  reader.onload = function (e) {
-    if (length === 0) {
-      console.error("can't stream chunk of length 0");
+  reader.onload = (e) => cb(btoa(e.target.result));
+  reader.onerror = () => cb(null);
+  reader.readAsBinaryString(file.slice(start, start + length));
+}
+
+function pumpStream(streamUuid) {
+  const s = activeStreams[streamUuid];
+  if (!s || s.reading) return;
+  if (s.ws.readyState !== WebSocket.OPEN) {
+    delete activeStreams[streamUuid];
+    return;
+  }
+  if (s.inflight >= STREAM_WINDOW) return;
+  if (s.ws.bufferedAmount > STREAM_MAX_BUFFERED) {
+    setTimeout(() => pumpStream(streamUuid), 50);
+    return;
+  }
+
+  const start = s.position;
+  const length = Math.min(STREAM_CHUNK_SIZE, s.share.size - start);
+  const close = start + length >= s.share.size;
+  s.position += length;
+  s.reading = true;
+
+  const sendChunk = (b64) => {
+    s.reading = false;
+    if (!activeStreams[streamUuid]) return; // aborted meanwhile
+    if (b64 === null) {
+      console.error("can't read file chunk for stream " + streamUuid);
+      delete activeStreams[streamUuid];
       return;
     }
-    const close = (start + length) === share.file.size;
-    ws.send(JSON.stringify({
+    s.inflight++;
+    s.ws.send(JSON.stringify({
       type: "chunk",
       uuid: streamUuid,
       close: close,
-      chunk: btoa(e.target.result)
+      chunk: b64
     }));
-    if (cb) cb(close);
+    if (close) {
+      delete activeStreams[streamUuid];
+      if (s.cb) s.cb();
+    } else {
+      pumpStream(streamUuid);
+    }
   };
-  const blob = share.file.slice(start, start + length);
-  reader.readAsBinaryString(blob);
+
+  if (length === 0) {
+    sendChunk(""); // empty file: a single closing chunk
+  } else {
+    readChunkAsBase64(s.share.file, start, length, sendChunk);
+  }
 }
 
 function streamShare(share, streamUuid, ws, cb) {
@@ -41,18 +91,25 @@ function streamShare(share, streamUuid, ws, cb) {
       chunk: btoa(share.content)
     }));
     if (cb) cb();
-  } else {
-    let position = 0;
-    function chunkStreamed(done) {
-      if (done) {
-        if (cb) cb();
-        return;
-      }
-      const start = position;
-      const length = Math.min(1024000, share.size - position);
-      position += length;
-      streamChunk(share, streamUuid, start, length, ws, chunkStreamed);
-    }
-    chunkStreamed(false);
+    return;
   }
+  activeStreams[streamUuid] = { share, ws, position: 0, inflight: 0, reading: false, cb };
+  pumpStream(streamUuid);
+}
+
+// Server consumed a chunk: we may send another one.
+function streamAck(streamUuid) {
+  const s = activeStreams[streamUuid];
+  if (!s) return;
+  s.inflight = Math.max(0, s.inflight - 1);
+  pumpStream(streamUuid);
+}
+
+// Downloader went away (or the socket died): stop reading the file.
+function streamAbort(streamUuid) {
+  delete activeStreams[streamUuid];
+}
+
+function streamAbortAll() {
+  Object.keys(activeStreams).forEach((uuid) => delete activeStreams[uuid]);
 }
diff --git a/spec/app_spec.rb b/spec/app_spec.rb
index 03fb875..3716bcb 100644
--- a/spec/app_spec.rb
+++ b/spec/app_spec.rb
@@ -47,3 +47,80 @@
   end
 
 end
+
+describe "WebSocket endpoint" do
+  let(:counts) { app.settings.connection_counts }
+  before { counts.clear }
+
+  def ws_headers(connection: 'Upgrade')
+    {
+      'HTTP_CONNECTION' => connection,
+      'HTTP_UPGRADE' => 'websocket',
+      'HTTP_SEC_WEBSOCKET_KEY' => 'dGhlIHNhbXBsZSBub25jZQ==',
+      'HTTP_SEC_WEBSOCKET_VERSION' => '13'
+    }
+  end
+
+  it "rejects non-websocket requests without leaking a connection slot" do
+    get "/ws"
+    expect(last_response.status).to eq(400)
+    expect(counts).to be_empty
+  end
+
+  it "releases the connection slot when the handshake fails" do
+    # Outside Thin there is no async callback: the handshake raises (like
+    # em-websocket's HandshakeError does) after the slot was taken.
+    60.times { get "/ws", {}, ws_headers(connection: 'upgrade') }
+    expect(last_response.status).to eq(400)
+    expect(counts).to be_empty
+    get "/ws", {}, ws_headers
+    expect(last_response.status).not_to eq(429)
+  end
+
+  it "normalizes a lowercase Connection header before the handshake" do
+    seen = nil
+    SinatraWebsocket::Connection.stubs(:from_env).with { |env, *| seen = env['HTTP_CONNECTION']; true }
+      .returns([400, {}, ['stubbed']])
+    get "/ws", {}, ws_headers(connection: 'keep-alive, upgrade')
+    expect(seen).to eq('Upgrade')
+  end
+
+  it "still enforces the per-IP limit for live connections" do
+    counts['127.0.0.1'] = app.settings.max_connections_per_ip
+    get "/ws", {}, ws_headers
+    expect(last_response.status).to eq(429)
+  end
+end
+
+describe "Downloads" do
+  let(:registry) { app.settings.registry.tap { |registry| registry.reset } }
+  let(:namespace) { registry.context_for("127.0.0.1").namespace_for(:default) }
+  let(:client) { DLCenter::Client.new(namespace).tap { |c| namespace.add_client(c) } }
+  let(:share) { DLCenter::Share.new(client, name: 'a"b/c.txt', size: 5, content_type: 'text/plain').tap { |s| client.add_share(s) } }
+
+  before do
+    # Complete the stream synchronously so the (non-EM) test scheduler finishes.
+    client.define_singleton_method(:send_msg) do |msg, params = {}|
+      next unless msg == :stream
+      stream = @streams[params[:uuid]]
+      stream.got_chunk("hello")
+      stream.drain_buffer
+      stream.close
+    end
+    EM.stubs(:next_tick).yields
+  end
+
+  it "serves a share with headers that disable proxy buffering" do
+    get "/share/#{share.uuid}"
+    expect(last_response).to be_ok
+    expect(last_response.headers['X-Accel-Buffering']).to eq('no')
+    expect(last_response.headers['Content-Length']).to eq('5')
+    expect(last_response.headers['Content-Disposition']).to eq('attachment; filename="a_b_c.txt"')
+    expect(last_response.body).to eq("hello")
+  end
+
+  it "returns 404 for an unknown share" do
+    get "/share/#{SecureRandom.uuid}"
+    expect(last_response.status).to eq(404)
+  end
+end
diff --git a/spec/thin_ipv6_host_spec.rb b/spec/thin_ipv6_host_spec.rb
new file mode 100644
index 0000000..96c7a0e
--- /dev/null
+++ b/spec/thin_ipv6_host_spec.rb
@@ -0,0 +1,28 @@
+require 'dlcenter/thin_ipv6_host'
+
+RSpec.describe DLCenter::ThinIPv6Host do
+  def parse(host)
+    request = Thin::Request.new
+    request.parse("GET / HTTP/1.1\r\nHost: #{host}\r\n\r\n")
+    request.env
+  end
+
+  it "keeps a bracketed IPv6 host and its port intact" do
+    env = parse("[::1]:55100")
+    expect(env['SERVER_NAME']).to eq("[::1]")
+    expect(env['SERVER_PORT']).to eq("55100")
+    expect { Rack::Lint.new(->(*) { [200, {}, []] }).call(env) }.not_to raise_error
+  end
+
+  it "defaults the port for a bracketed IPv6 host without one" do
+    env = parse("[::1]")
+    expect(env['SERVER_NAME']).to eq("[::1]")
+    expect(env['SERVER_PORT']).to eq("80")
+  end
+
+  it "leaves regular hosts alone" do
+    env = parse("localhost:55100")
+    expect(env['SERVER_NAME']).to eq("localhost")
+    expect(env['SERVER_PORT']).to eq("55100")
+  end
+end
diff --git a/spec/ws_client_spec.rb b/spec/ws_client_spec.rb
new file mode 100644
index 0000000..1022055
--- /dev/null
+++ b/spec/ws_client_spec.rb
@@ -0,0 +1,159 @@
+require 'json'
+require 'dlcenter'
+
+# Minimal stand-in for SinatraWebsocket::Connection (em-websocket 0.3.x API):
+# single-slot callbacks, `send`, and `close_websocket` (there is no `close`).
+class FakeWS
+  attr_reader :sent, :closed
+  def initialize
+    @sent = []
+    @closed = false
+  end
+  def onopen(&blk);    @onopen = blk;    end
+  def onmessage(&blk); @onmessage = blk; end
+  def onclose(&blk);   @onclose = blk;   end
+  def send(msg);       @sent << JSON.parse(msg, symbolize_names: true); end
+  def close_websocket(*) @closed = true; end
+  def open;            @onopen.call;     end
+  def message(msg);    @onmessage.call(msg.is_a?(String) ? msg : msg.to_json); end
+  def close;           @onclose.call;    end
+end
+
+class FastHeartbeatClient < DLCenter::WSClient
+  HEARTBEAT_INTERVAL = 0.05
+  HEARTBEAT_TIMEOUT = 0.05
+end
+
+class FakeConnection
+  attr_accessor :outbound
+  def initialize(outbound = 0) @outbound = outbound end
+  def get_outbound_data_size; @outbound; end
+end
+
+RSpec.describe DLCenter::WSClient do
+  let(:namespace) { DLCenter::Namespace.new(:default) }
+  let(:ws) { FakeWS.new }
+  let(:closed_calls) { [] }
+  let(:client) { DLCenter::WSClient.new(namespace, ws) { closed_calls << :closed } }
+  let(:uuid) { SecureRandom.uuid }
+
+  before { namespace.add_client(client) }
+
+  def em(timeout: 2)
+    EM.run do
+      EM.add_timer(timeout) { EM.stop }
+      yield
+    end
+  end
+
+  it "removes itself from the namespace and reports the close" do
+    ws.close
+    expect(namespace.clients).not_to include(client)
+    expect(closed_calls).to eq([:closed])
+  end
+
+  it "closes the downloads it was feeding when it disconnects" do
+    share = DLCenter::Share.new(client, name: "file", size: 3)
+    client.add_share(share)
+    out = StringIO.new
+    stream = nil
+    em do
+      stream = share.content(out)
+      ws.close
+      EM.next_tick { EM.stop }
+    end
+    expect(stream).to be_closed
+    expect(out).to be_closed
+  end
+
+  it "closes the socket with close_websocket when the heartbeat times out" do
+    client = FastHeartbeatClient.new(namespace, ws)
+    em do
+      ws.open
+      EM.add_timer(0.5) { EM.stop }
+    end
+    expect(ws.sent.map { |m| m[:type] }).to include(:ping.to_s)
+    expect(ws.closed).to eq(true)
+  end
+
+  it "keeps the connection when the client answers pings" do
+    client = FastHeartbeatClient.new(namespace, ws)
+    em do
+      ws.open
+      EM.add_periodic_timer(0.02) { ws.message(type: 'pong') }
+      EM.add_timer(0.4) { EM.stop }
+    end
+    expect(ws.closed).to eq(false)
+    client.stop_heartbeat
+  end
+
+  it "does not crash the reactor when a message handler raises" do
+    client.stubs(:handle_register_share).raises(RuntimeError, "boom")
+    expect { ws.message(type: 'register_share', name: 'x') }.not_to raise_error
+  end
+
+  describe "streaming with flow control" do
+    let(:share) { DLCenter::Share.new(client, name: "file", size: 6) }
+    let(:out) { StringIO.new }
+
+    before { client.add_share(share) }
+
+    def chunk_msg(stream_uuid, data, close: false)
+      { type: 'chunk', uuid: stream_uuid, chunk: Base64.strict_encode64(data), close: close }
+    end
+
+    it "asks the sender for a stream and acks chunks once drained" do
+      stream = nil
+      em do
+        stream = share.content(out)
+        ws.message(chunk_msg(stream.uuid, "abc"))
+        EM.add_timer(0.1) { EM.stop }
+      end
+      stream_msg = ws.sent.find { |m| m[:type] == 'stream' }
+      expect(stream_msg).to include(uuid: stream.uuid, share: share.uuid)
+      expect(ws.sent).to include(type: 'ack', uuid: stream.uuid)
+      expect(out.string).to eq("abc")
+    end
+
+    it "withholds the ack while the downloader's socket is backed up" do
+      connection = FakeConnection.new(DLCenter::Streamer::HIGH_WATER * 2)
+      stream = nil
+      em do
+        stream = share.content(out, connection: connection)
+        ws.message(chunk_msg(stream.uuid, "abc"))
+        EM.add_timer(0.2) do
+          expect(ws.sent).not_to include(type: 'ack', uuid: stream.uuid)
+          connection.outbound = 0
+        end
+        EM.add_timer(0.4) { EM.stop }
+      end
+      expect(ws.sent).to include(type: 'ack', uuid: stream.uuid)
+    end
+
+    it "closes the download on the final chunk" do
+      stream = nil
+      em do
+        stream = share.content(out)
+        ws.message(chunk_msg(stream.uuid, "abc"))
+        ws.message(chunk_msg(stream.uuid, "def", close: true))
+        EM.add_timer(0.1) { EM.stop }
+      end
+      expect(out.string).to eq("abcdef")
+      expect(out).to be_closed
+      expect(client.active_streams).to be_empty
+    end
+
+    it "tells the sender to stop when the downloader disconnects" do
+      closer = EM::DefaultDeferrable.new
+      stream = nil
+      em do
+        stream = share.content(out, closer: closer)
+        closer.succeed
+        EM.next_tick { EM.stop }
+      end
+      expect(ws.sent).to include(type: 'stream_close', uuid: stream.uuid)
+      expect(stream).to be_closed
+      expect(client.active_streams).to be_empty
+    end
+  end
+end