[PATCH v5 http-server 0/2] fix pveproxy OOM in websocket and spice proxy handlers

[PATCH v5 http-server 0/2] fix pveproxy OOM in websocket and spice proxy handlers

[Kefu Chai]

see v2's cover letter [1] for the problem description and the approach.

v4 does not fix the reported bug. On a TLS handle, AnyEvent keeps
reading even after on_read is cleared, so the websocket client side
kept filling rbuf until it hit rbuf_max, and the connection was dropped
with ENOSPC. That is the direction PDM migration sends its data in.

Patch 1 is v4 with the fixes below. Patch 2 is new: when one side of a
proxied connection closes, the proxy dropped the data still queued for
the other side. v4 tried to address this with the drain in on_eof, but
the data is lost in client_do_disconnect(), which shuts the socket down
before its wbuf is written. Patch 2 waits for the queued data to be
written before disconnecting. It uses the same timeout rule as patch 1:
60 seconds when waiting for the client, none when waiting for the
backend. It is a separate patch because master has the same problem,
and it can be reviewed on its own.

Changes to patch 1 since v4 [2]:

* call stop_read() when pausing, so the pause also works on the TLS
  client handle.

* set a 60 second write timeout when the client stops reading. Without
  it, such a client kept the connection open forever, because the proxy
  handles have no timeout and a paused handle does not notice EOF.
  There is no timeout when the backend stops reading. The backend is a
  local process, and closing its connection early drops the data still
  queued for it. For a remote disk import in raw+size format, dd then
  reaches EOF and the import succeeds with the end of the disk missing.

* on a WebSocket Close frame, disconnect once the queued data is
  written, the same way finish_response() does, instead of calling
  push_shutdown(). push_shutdown() only half-closed the socket and
  dropped a paused backend reader, so the backend was never read again.

* drop the rbuf drain in on_eof, together with handle_proxy_eof(). A
  paused handle has no read watcher, so EOF only shows up after the
  reader has resumed and emptied rbuf, and there is never anything left
  to drain. This also removes the eval split added in v4.

* do not call the previous on_drain twice when resuming, since
  on_drain() already calls it when restoring it.

* use abort_request() (was handle_proxy_error()) for the accept-time
  client handle as well, define the 640 KB limit once as
  $limit_proxy_wbuf, and move the push-and-pause code of the four proxy
  readers into proxy_forward().

I tested master, v4 and this series with an out-of-tree setup (not part
of this series): the real server with TLS and a stub rest_handler, a
websocket client, and a plain TCP backend.

* with the client sending 256 MB and the backend reading at 32 MB/s,
  master grew by 246 MB and v4 dropped the connection after about
  7 MB. v5 finished with the data intact and no memory growth.

* when the client stopped reading, v5 disconnected after 60 seconds,
  and v4 kept the connection open past the 90 second limit of the test.
  When the backend stopped reading, v5 kept the connection open with
  about 1 MB of memory growth. master buffered whatever arrived in both
  cases (1 GB in the test).

* after a Close frame sent while paused, v5 flushed the queued data and
  closed both sides. v4 closed too, but only because the client gave up
  on a TLS error.

* when the backend closed after sending its data, master lost 252 MB
  of 256 MB. With 64 MB, v4 lost up to 576 KB in 3 of 5 runs and patch
  1 alone up to 624 KB in 4 of 7 runs, and with patch 2 all 7 runs were
  complete. When the client closed without a Close frame, patch 1 alone
  lost up to 691 KB in 3 of 5 runs, and with patch 2 all 7 runs were
  complete. When the client did not read after the backend closed,
  patch 2 disconnected after 60 seconds.

* I also replayed an offline remote disk migration through the real
  proxmox-websocket-tunnel, with stubs for the mtunnel worker and for
  pvesm import. The tunnel ends each forwarded connection with a Close
  frame, so both master and v5 delivered the full disk. With the
  import pausing for 75 seconds when 2 MB were left, v5 still delivered
  the full disk. A version that also timed out on the backend closed
  the connection instead, and the import ended 1.9 MB short without an
  error, which is why v5 has no timeout there.

[1] https://lore.proxmox.com/pve-devel/20260413125650.2569621-1-k.chai@proxmox.com/
[2] https://lore.proxmox.com/pve-devel/20260617122905.3822836-1-k.chai@proxmox.com/

Kefu Chai (2):
  fix #7483: apiserver: add backpressure to proxy handlers
  apiserver: flush queued data before closing a proxied connection

 src/PVE/APIServer/AnyEvent.pm | 189 +++++++++++++++++++++-------------
 1 file changed, 119 insertions(+), 70 deletions(-)

-- 
2.47.3

[PATCH v5 http-server 1/2] fix #7483: apiserver: add backpressure to proxy handlers

[Kefu Chai]

When pveproxy forwards a WebSocket or SPICE connection, it reads from
one side as fast as data arrives, even if the other side is slower. The
extra data piles up in memory and pveproxy can get OOM-killed. A PDM
cross-cluster migration to LVM-thin hits this easily, because the
target is slow while it zeroes newly allocated extents on first write.

Setting wbuf_max on the backend handle does not prevent this. AnyEvent
only checks it before the first write watcher is installed. Even if the
check worked, it would drop the connection with ENOSPC instead of
slowing it down.

Move the read-pausing code from response_stream() into a new
apply_read_backpressure() helper and use it in both proxies. It stops
reading from one side when more than 640 KB is queued for the other
side, and starts reading again from on_drain. The helper also calls
stop_read(). Clearing on_read is not enough for TLS handles, because
AnyEvent keeps reading them into rbuf until it hits rbuf_max.

While reads are paused, nothing watches that side of the connection,
and the proxy handles have no timeout. If the client is the side that
stopped reading, set a 60 second write timeout, so the connection gets
closed instead of staying open forever. This also covers a stalled
download in response_stream(). There is no timeout when the backend
stops reading. The backend is a local process, and closing its
connection early would drop the data still queued for it. For a remote
disk import, the import would then succeed with the end of the disk
missing.

When a WebSocket Close frame arrives, disconnect once the queued data
is written, like finish_response() does. Do not use push_shutdown()
there. It only half-closes the socket, and it overwrites the on_drain
callback that resumes a paused backend reader, so the backend would
never be read again.

The on_eof and on_error callbacks of the proxy and client handles all
did the same logging and disconnect. Move that into abort_request().

Signed-off-by: Kefu Chai <k.chai@proxmox.com>
---
 src/PVE/APIServer/AnyEvent.pm | 152 ++++++++++++++++++----------------
 1 file changed, 82 insertions(+), 70 deletions(-)

diff --git a/src/PVE/APIServer/AnyEvent.pm b/src/PVE/APIServer/AnyEvent.pm
index 915d678..826e1ff 100644
--- a/src/PVE/APIServer/AnyEvent.pm
+++ b/src/PVE/APIServer/AnyEvent.pm
@@ -52,6 +52,10 @@ use PVE::APIServer::Utils;
 my $limit_max_headers = 64;
 my $limit_max_header_size = 8 * 1024;
 my $limit_max_post = 512 * 1024;
+# proxy backpressure: pause reading once this much is queued for the peer, and
+# give up on a client that takes no data for this long
+my $limit_proxy_wbuf = 640 * 1024;
+my $backpressure_stall_timeout = 60;
 
 my $known_methods = {
     GET => 1,
@@ -206,6 +210,55 @@ sub finish_response {
     }
 }
 
+# pause reads on $read_hdl until $write_hdl's wbuf drains. The reader to resume
+# is taken from the handle rather than passed in, so it need not reference
+# itself (a self-referential closure is a cycle that would pin $reqstate).
+# Only pass $timeout if $write_hdl is the client: closing a stalled backend
+# would drop data it has not read yet, e.g. the end of a disk import.
+sub apply_read_backpressure {
+    my ($read_hdl, $write_hdl, $timeout) = @_;
+
+    my $on_read_cb = $read_hdl->{on_read};
+    return if !$on_read_cb;
+
+    $read_hdl->on_read();
+    # on_read() alone keeps the read watcher on TLS handles, see _drain_rbuf
+    $read_hdl->stop_read();
+    if ($timeout) {
+        $write_hdl->wtimeout_reset();
+        $write_hdl->wtimeout($timeout);
+    }
+    my $prev_on_drain = $write_hdl->{on_drain};
+    $write_hdl->on_drain(sub {
+        my ($wrhdl) = @_;
+        $wrhdl->wtimeout(0) if $timeout;
+        # restoring the previous on_drain also invokes it, as wbuf is empty here
+        $wrhdl->on_drain($prev_on_drain);
+        $read_hdl->on_read($on_read_cb);
+    });
+}
+
+# forward $data to $write_hdl, and pause $read_hdl if too much is queued there.
+# Returns true if reads were paused.
+sub proxy_forward {
+    my ($read_hdl, $write_hdl, $data, $timeout) = @_;
+
+    $write_hdl->push_write($data);
+    return 0 if length($write_hdl->{wbuf}) <= $limit_proxy_wbuf;
+
+    apply_read_backpressure($read_hdl, $write_hdl, $timeout);
+    return 1;
+}
+
+sub abort_request {
+    my ($self, $reqstate, $message) = @_;
+    eval {
+        $self->log_aborted_request($reqstate, $message);
+        $self->client_do_disconnect($reqstate);
+    };
+    if (my $err = $@) { syslog('err', $err); }
+}
+
 sub response_stream {
     my ($self, $reqstate, $stream_fh) = @_;
 
@@ -214,16 +267,13 @@ sub response_stream {
 
     my $buf_size = 4 * 1024 * 1024;
 
-    my $on_read;
-    $on_read = sub {
+    my $on_read = sub {
         my ($hdl) = @_;
         my $reqhdl = $reqstate->{hdl};
         return if !$reqhdl;
 
         my $wbuf_len = length($reqhdl->{wbuf});
         my $rbuf_len = length($hdl->{rbuf});
-        # TODO: Take into account $reqhdl->{wbuf_max} ? Right now
-        # that's unbounded, so just assume $buf_size
         my $to_read = $buf_size - $wbuf_len;
         $to_read = $rbuf_len if $rbuf_len < $to_read;
         if ($to_read > 0) {
@@ -241,20 +291,9 @@ sub response_stream {
 
         # apply backpressure so we don't accept any more data into buffer if the client isn't
         # downloading fast enough. Note: read_size can double upon read, and we also need to account
-        # for one more read after# start_read, so multiply by 4
+        # for one more read after start_read, so multiply by 4
         if ($rbuf_len + $hdl->{read_size} * 4 > $buf_size) {
-            # stop reading until write buffer is empty
-            $hdl->on_read();
-            my $prev_on_drain = $reqhdl->{on_drain};
-            $reqhdl->on_drain(sub {
-                my ($wrhdl) = @_;
-                # on_drain called because write buffer is empty, continue reading
-                $hdl->on_read($on_read);
-                if ($prev_on_drain) {
-                    $wrhdl->on_drain($prev_on_drain);
-                    $prev_on_drain->($wrhdl);
-                }
-            });
+            apply_read_backpressure($hdl, $reqhdl, $backpressure_stall_timeout);
         }
     };
 
@@ -276,16 +315,10 @@ sub response_stream {
                 }
             };
             if (my $err = $@) { syslog('err', "$err"); }
-            $on_read = undef;
         },
         on_error => sub {
             my ($hdl, $fatal, $message) = @_;
-            eval {
-                $self->log_aborted_request($reqstate, $message);
-                $self->client_do_disconnect($reqstate);
-            };
-            if (my $err = $@) { syslog('err', "$err"); }
-            $on_read = undef;
+            $self->abort_request($reqstate, $message);
         },
     );
 }
@@ -584,37 +617,29 @@ sub websocket_proxy {
             $reqstate->{proxyhdl} = AnyEvent::Handle->new(
                 fh => $fh,
                 rbuf_max => $max_payload_size,
-                wbuf_max => $max_payload_size * 5,
                 timeout => 5,
                 on_eof => sub {
                     my ($hdl) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', $err); }
+                    $self->abort_request($reqstate);
                 },
                 on_error => sub {
                     my ($hdl, $fatal, $message) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate, $message);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', "$err"); }
+                    $self->abort_request($reqstate, $message);
                 },
             );
 
             my $proxyhdlreader = sub {
                 my ($hdl) = @_;
 
+                my $clienthdl = $reqstate->{hdl};
+                return if !$clienthdl;
+
                 my $len = length($hdl->{rbuf});
                 my $data =
                     substr($hdl->{rbuf}, 0, $len > $max_payload_size ? $max_payload_size : $len,
                         '');
 
-                my $string = $encode->(\$data);
-
-                $reqstate->{hdl}->push_write($string) if $reqstate->{hdl};
+                proxy_forward($hdl, $clienthdl, $encode->(\$data), $backpressure_stall_timeout);
             };
 
             my $hdlreader = sub {
@@ -672,7 +697,8 @@ sub websocket_proxy {
                     }
 
                     if ($opcode == 1 || $opcode == 2) {
-                        $reqstate->{proxyhdl}->push_write($payload) if $reqstate->{proxyhdl};
+                        my $proxyhdl = $reqstate->{proxyhdl};
+                        return if $proxyhdl && proxy_forward($hdl, $proxyhdl, $payload);
                     } elsif ($opcode == 8) {
                         my $statuscode = unpack("n", $payload);
                         $self->dprint("websocket received close. status code: '$statuscode'");
@@ -681,7 +707,10 @@ sub websocket_proxy {
 
                             $proxyhdl->push_shutdown();
                         }
-                        $hdl->push_shutdown();
+                        # tear down once the queued data is flushed. push_shutdown()
+                        # would only half-close, and drop a backend reader paused
+                        # in on_drain, so the backend would never be read again.
+                        $hdl->on_drain(sub { $self->abort_request($reqstate) });
                     } elsif ($opcode == 9) {
                         # ping received, schedule pong
                         $reqstate->{hdl}->push_write($encode->(\$payload, "\x8A"))
@@ -700,8 +729,6 @@ sub websocket_proxy {
             $reqstate->{proxyhdl}->on_read($proxyhdlreader);
             $reqstate->{hdl}->on_read($hdlreader);
 
-            # todo: use stop_read/start_read if write buffer grows to much
-
             # FIXME: remove protocol in PVE/PMG 8.x
             #
             # for backwards, compatibility,  we have to reply with the websocket
@@ -1098,7 +1125,6 @@ sub handle_spice_proxy_request {
         }
 
         $reqstate->{hdl}->timeout(0);
-        $reqstate->{hdl}->wbuf_max(64 * 10 * 1024);
 
         my $remhost = $remip ? $remip : "localhost";
         my $remport = $remip ? 3128 : $spiceport;
@@ -1108,47 +1134,43 @@ sub handle_spice_proxy_request {
                 or die "connect to '$remhost:$remport' failed: $!";
 
             $self->dprint("CONNECTed to '$remhost:$remport'");
+
             $reqstate->{proxyhdl} = AnyEvent::Handle->new(
                 fh => $fh,
                 rbuf_max => 64 * 1024,
-                wbuf_max => 64 * 10 * 1024,
                 timeout => 5,
                 on_eof => sub {
                     my ($hdl) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', $err); }
+                    $self->abort_request($reqstate);
                 },
                 on_error => sub {
                     my ($hdl, $fatal, $message) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate, $message);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', "$err"); }
+                    $self->abort_request($reqstate, $message);
                 },
             );
 
             my $proxyhdlreader = sub {
                 my ($hdl) = @_;
 
+                my $clienthdl = $reqstate->{hdl};
+                return if !$clienthdl;
+
                 my $len = length($hdl->{rbuf});
                 my $data = substr($hdl->{rbuf}, 0, $len, '');
 
-                #print "READ1 $len\n";
-                $reqstate->{hdl}->push_write($data) if $reqstate->{hdl};
+                proxy_forward($hdl, $clienthdl, $data, $backpressure_stall_timeout);
             };
 
             my $hdlreader = sub {
                 my ($hdl) = @_;
 
+                my $proxyhdl = $reqstate->{proxyhdl};
+                return if !$proxyhdl;
+
                 my $len = length($hdl->{rbuf});
                 my $data = substr($hdl->{rbuf}, 0, $len, '');
 
-                #print "READ0 $len\n";
-                $reqstate->{proxyhdl}->push_write($data) if $reqstate->{proxyhdl};
+                proxy_forward($hdl, $proxyhdl, $data);
             };
 
             my $proto = $reqstate->{proto} ? $reqstate->{proto}->{str} : 'HTTP/1.0';
@@ -1158,8 +1180,6 @@ sub handle_spice_proxy_request {
                 $reqstate->{proxyhdl}->on_read($proxyhdlreader);
                 $reqstate->{hdl}->on_read($hdlreader);
 
-                # todo: use stop_read/start_read if write buffer grows to much
-
                 # a response must be followed by an empty line
                 my $res = "$proto 200 OK\015\012\015\012";
                 $reqstate->{hdl}->push_write($res);
@@ -1978,19 +1998,11 @@ sub accept_connections {
                 linger => 0, # avoid problems with ssh - really needed ?
                 on_eof => sub {
                     my ($hdl) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', $err); }
+                    $self->abort_request($reqstate);
                 },
                 on_error => sub {
                     my ($hdl, $fatal, $message) = @_;
-                    eval {
-                        $self->log_aborted_request($reqstate, $message);
-                        $self->client_do_disconnect($reqstate);
-                    };
-                    if (my $err = $@) { syslog('err', "$err"); }
+                    $self->abort_request($reqstate, $message);
                 },
             );
             $handle_creation = 0;
-- 
2.47.3

[PATCH v5 http-server 2/2] apiserver: flush queued data before closing a proxied connection

[Kefu Chai]

When one side of a proxied WebSocket or SPICE connection closes, the
proxy disconnects right away and drops the data still queued for the
other side. client_do_disconnect() shuts the socket down before its
wbuf is written, and the client handle has linger set to 0. With the
backpressure limit this is up to 640 KB at the end of a transfer.

On EOF from either side, stop reading from the other side and
disconnect from its on_drain callback once the queued data is written.
Also clear the on_drain callback of the closed side, so a paused reader
of the other side is not resumed. When waiting for the client, the 60
second write timeout applies, so a client that stops reading does not
keep the connection open. When waiting for the backend, there is no
timeout, for the same reason as in apply_read_backpressure().

Signed-off-by: Kefu Chai <k.chai@proxmox.com>
---
 src/PVE/APIServer/AnyEvent.pm | 41 +++++++++++++++++++++++++++++++++--
 1 file changed, 39 insertions(+), 2 deletions(-)

diff --git a/src/PVE/APIServer/AnyEvent.pm b/src/PVE/APIServer/AnyEvent.pm
index 826e1ff..c8266df 100644
--- a/src/PVE/APIServer/AnyEvent.pm
+++ b/src/PVE/APIServer/AnyEvent.pm
@@ -259,6 +259,31 @@ sub abort_request {
     if (my $err = $@) { syslog('err', $err); }
 }
 
+# on_eof body for proxied connections: stop forwarding, and disconnect once
+# $peer_hdl has written out what is still queued for it.
+sub disconnect_after_drain {
+    my ($self, $reqstate, $hdl, $peer_hdl, $timeout) = @_;
+
+    if (!$peer_hdl) {
+        $self->abort_request($reqstate);
+        return;
+    }
+
+    # drop a paused $peer_hdl reader, so it is not resumed
+    $hdl->on_drain(undef);
+    $hdl->wtimeout(0);
+    $peer_hdl->on_read();
+    $peer_hdl->stop_read();
+
+    # only pass $timeout if $peer_hdl is the client, see apply_read_backpressure
+    if ($timeout) {
+        $peer_hdl->wtimeout_reset();
+        $peer_hdl->wtimeout($timeout);
+    }
+    # invoked right away if nothing is queued
+    $peer_hdl->on_drain(sub { $self->abort_request($reqstate) });
+}
+
 sub response_stream {
     my ($self, $reqstate, $stream_fh) = @_;
 
@@ -620,7 +645,9 @@ sub websocket_proxy {
                 timeout => 5,
                 on_eof => sub {
                     my ($hdl) = @_;
-                    $self->abort_request($reqstate);
+                    $self->disconnect_after_drain(
+                        $reqstate, $hdl, $reqstate->{hdl}, $backpressure_stall_timeout,
+                    );
                 },
                 on_error => sub {
                     my ($hdl, $fatal, $message) = @_;
@@ -728,6 +755,10 @@ sub websocket_proxy {
             $reqstate->{proxyhdl}->timeout(0);
             $reqstate->{proxyhdl}->on_read($proxyhdlreader);
             $reqstate->{hdl}->on_read($hdlreader);
+            $reqstate->{hdl}->on_eof(sub {
+                my ($hdl) = @_;
+                $self->disconnect_after_drain($reqstate, $hdl, $reqstate->{proxyhdl});
+            });
 
             # FIXME: remove protocol in PVE/PMG 8.x
             #
@@ -1141,7 +1172,9 @@ sub handle_spice_proxy_request {
                 timeout => 5,
                 on_eof => sub {
                     my ($hdl) = @_;
-                    $self->abort_request($reqstate);
+                    $self->disconnect_after_drain(
+                        $reqstate, $hdl, $reqstate->{hdl}, $backpressure_stall_timeout,
+                    );
                 },
                 on_error => sub {
                     my ($hdl, $fatal, $message) = @_;
@@ -1179,6 +1212,10 @@ sub handle_spice_proxy_request {
                 $reqstate->{proxyhdl}->timeout(0);
                 $reqstate->{proxyhdl}->on_read($proxyhdlreader);
                 $reqstate->{hdl}->on_read($hdlreader);
+                $reqstate->{hdl}->on_eof(sub {
+                    my ($hdl) = @_;
+                    $self->disconnect_after_drain($reqstate, $hdl, $reqstate->{proxyhdl});
+                });
 
                 # a response must be followed by an empty line
                 my $res = "$proto 200 OK\015\012\015\012";
-- 
2.47.3