From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id 4CE901FF0B3 for ; Fri, 11 Sep 2026 16:19:08 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 2B66A2153C; Fri, 11 Sep 2026 16:18:57 +0200 (CEST) From: Kefu Chai To: pve-devel@lists.proxmox.com Subject: [PATCH v5 http-server 1/2] fix #7483: apiserver: add backpressure to proxy handlers Date: Fri, 11 Sep 2026 22:18:45 +0800 Message-ID: <20260911141846.9888-2-k.chai@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20260911141846.9888-1-k.chai@proxmox.com> References: <20260911141846.9888-1-k.chai@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1789136321865 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.200 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: 65BND5U6ZDBSHQJUNCDR5DZRJJGB52ST X-Message-ID-Hash: 65BND5U6ZDBSHQJUNCDR5DZRJJGB52ST X-MailFrom: k.chai@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: 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 --- 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