Fix WebSocket disconnections, server crashes and big-file downloads

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:

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