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

[Kefu Chai <k.chai@proxmox.com>]

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