Merge 5bbb08f into 67f295b

src/common/ngx_wasm_socket_tcp_readers.c

@@ -623,9 +623,9 @@ ngx_wasm_read_http_response(ngx_buf_t *src, ngx_chain_t *buf_in, ssize_t bytes,
             buf->last = ngx_cpymem(buf->last, p, chunk_len);
         }
 
+#if 0
         ngx_wasm_assert((size_t) ngx_buf_size(buf) == in_ctx->body_len);
 
-#if 0
         dd("body_len: %ld", in_ctx->body_len);
 
         for (ll = in_ctx->body; ll; ll = ll->next) {

t/03-proxy_wasm/hfuncs/102-proxy_send_local_response.t

@@ -143,6 +143,7 @@ qr/.*? on_request_headers, 2 headers, .*
 .*? testing in "RequestHeaders", .*
 .*? on_response_headers, 5 headers, .*
 .*? on_response_body, 0 bytes, eof: true, .*
+.*? on_done.*
 .*? on_log.*/
 --- no_error_log
 [error]
@@ -248,6 +249,7 @@ qr/.*? on_request_headers, 2 headers, .*
 .*? testing in "RequestHeaders", .*
 .*? on_response_headers, 16 headers, .*
 .*? on_response_body, 0 bytes, eof: true, .*
+.*? on_done.*
 .*? on_log.*/
 --- no_error_log
 [error]
@@ -467,7 +469,9 @@ qr/.*? on_request_headers, \d+ headers.*
 .*? on_response_headers, \d+ headers.*
 .*? on_response_body, \d+ bytes, eof: true.*
 .*? on_response_body, \d+ bytes, eof: true.*
+.*? on_done.*
 .*? on_log.*
+.*? on_done.*
 .*? on_log.*/
 --- no_error_log
 [error]

t/04-openresty/ffi/103-proxy_wasm_attach.t

@@ -105,6 +105,7 @@ qr/^[^#]*#0 on_vm_start[^#]*
 #\d+ on_response_headers, 5 headers[^#]*
 #\d+ on_response_body, 3 bytes, eof: false[^#]*
 #\d+ on_response_body, 0 bytes, eof: true[^#]*
+#\d+ on_done[^#]*
 #\d+ on_log[^#]*$/
 --- error_log eval
 qr/\[error\] .*? previous plan already attached/
@@ -156,6 +157,7 @@ qr/^[^#]*#0 on_vm_start[^#]*
 #\d+ on_response_headers, 5 headers[^#]*
 #\d+ on_response_body, 3 bytes, eof: false[^#]*
 #\d+ on_response_body, 0 bytes, eof: true[^#]*
+#\d+ on_done[^#]*
 #\d+ on_log[^#]*$/
 --- no_error_log
 [error]
@@ -207,6 +209,7 @@ qr/^[^#]*#0 on_vm_start[^#]*
 #\d+ on_response_headers, 5 headers[^#]*
 #\d+ on_response_body, 3 bytes, eof: false[^#]*
 #\d+ on_response_body, 0 bytes, eof: true[^#]*
+#\d+ on_done[^#]*
 #\d+ on_log[^#]*$/
 --- no_error_log
 [error]
@@ -258,6 +261,7 @@ qr/^[^#]*#0 on_vm_start[^#]*
 #\d+ on_response_headers, 5 headers[^#]*
 #\d+ on_response_body, 3 bytes, eof: false[^#]*
 #\d+ on_response_body, 0 bytes, eof: true[^#]*
+#\d+ on_done[^#]*
 #\d+ on_log[^#]*$/
 --- no_error_log
 [error]

t/05-others/010-client_connection_abort.t

@@ -0,0 +1,128 @@
+# vim:set ft= ts=4 sts=4 sw=4 et fdm=marker:
+
+use strict;
+use lib '.';
+use t::TestWasm;
+
+plan tests => repeat_each() * (blocks() * 4);
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: client connection abort - wasm_call
+Nginx response filters are not invoked on the produced HTTP 499 response.
+--- skip_eval: 4: $ENV{TEST_NGINX_USE_HUP} == 1 or $::nginxV !~ m/debug/
+--- wasm_modules: ngx_rust_tests
+--- tcp_listen: $TEST_NGINX_UNIX_SOCKET
+--- tcp_reply eval
+sub {
+    return ["HTTP/1.1 200 OK\r\n",
+            "Connection: close\r\n",
+            "Content-Length: 0\r\n",
+            "\r\n"];
+}
+--- config
+    location /t {
+        wasm_call content ngx_rust_tests log_notice_hello;
+        wasm_call header_filter ngx_rust_tests log_notice_hello;
+        wasm_call body_filter ngx_rust_tests log_notice_hello;
+        proxy_pass http://unix:$TEST_NGINX_UNIX_SOCKET:/;
+    }
+--- shutdown
+--- ignore_response
+--- grep_error_log eval: qr/(client prematurely closed|(wasm ops calling .*? in .*? phase))/
+--- grep_error_log_out eval
+qr/wasm ops calling "ngx_rust_tests\.log_notice_hello" in "content" phase
+client prematurely closed\Z/
+--- no_error_log
+header_filter
+body_filter
+[error]
+
+
+
+=== TEST 2: client connection abort - chained filters
+Filters on aborted client connections do not execute response phases.
+--- skip_eval: 4: $ENV{TEST_NGINX_USE_HUP} == 1
+--- wasm_modules: on_phases
+--- tcp_listen: $TEST_NGINX_UNIX_SOCKET
+--- tcp_reply eval
+sub {
+    return ["HTTP/1.1 200 OK\r\n",
+            "Connection: close\r\n",
+            "Content-Length: 0\r\n",
+            "\r\n"];
+}
+--- config
+    location /t {
+        proxy_wasm on_phases;
+        proxy_wasm on_phases;
+        proxy_pass http://unix:$TEST_NGINX_UNIX_SOCKET:/;
+    }
+--- shutdown
+--- ignore_response
+--- grep_error_log eval: qr/\[info\] .*? ((client prematurely closed)|on_request_headers|on_response_headers|on_log|on_done|on_http_call_response.*)/
+--- grep_error_log_out eval
+qr/\A\[info\] .*? on_request_headers
+\[info\] .*? on_request_headers
+\[info\] .*? reported that client prematurely closed
+\[info\] .*? on_done
+\[info\] .*? on_log
+\[info\] .*? on_done
+\[info\] .*? on_log\Z/
+--- no_error_log
+on_response
+[error]
+[crit]
+
+
+
+=== TEST 3: client connection abort - chained filters with HTTP dispatch
+Filters with pending dispatch calls on aborted client connections await for
+response.
+--- skip_eval: 4: $ENV{TEST_NGINX_USE_HUP} == 1
+--- load_nginx_modules: ngx_http_echo_module
+--- wasm_modules: hostcalls
+--- http_config eval
+qq{
+    server {
+        listen unix:$ENV{TEST_NGINX_UNIX_SOCKET};
+
+        location / {
+            return 200;
+        }
+
+        location /sleep {
+            echo_sleep 0.3;
+            echo_status 201;
+        }
+    }
+}
+--- config
+    location /t {
+        proxy_wasm hostcalls 'test=/t/dispatch_http_call \
+                              host=unix:$TEST_NGINX_UNIX_SOCKET \
+                              path=/';
+        proxy_wasm hostcalls 'test=/t/dispatch_http_call \
+                              host=unix:$TEST_NGINX_UNIX_SOCKET \
+                              path=/sleep';
+        proxy_pass http://unix:$TEST_NGINX_UNIX_SOCKET:/;
+    }
+--- shutdown
+--- ignore_response
+--- grep_error_log eval: qr/\[info\] .*? ((client prematurely closed)|on_request_headers|on_response_headers|on_log|on_done|on_http_call_response.*)/
+--- grep_error_log_out eval
+qr/\A\[info\] .*? on_request_headers
+\[info\] .*? on_http_call_response \(id: \d+, status: 200[^*].*
+\[info\] .*? on_request_headers
+\[info\] .*? on_http_call_response \(id: \d+, status: 201[^*].*
+\[info\] .*? reported that client prematurely closed
+\[info\] .*? on_done
+\[info\] .*? on_log
+\[info\] .*? on_done
+\[info\] .*? on_log\Z/
+--- no_error_log
+[error]
+[crit]
+[emerg]

t/05-others/011-upstream_connection_abort.t

@@ -0,0 +1,84 @@
+# vim:set ft= ts=4 sts=4 sw=4 et fdm=marker:
+
+use strict;
+use lib '.';
+use t::TestWasm;
+
+skip_valgrind();
+
+plan tests => repeat_each() * (blocks() * 4);
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: upstream connection abort - wasm_call
+Calls on aborted upstream connections execute response phases on the produced
+HTTP 502 response.
+--- skip_eval: 4: $ENV{TEST_NGINX_USE_HUP} == 1 or $::nginxV !~ m/debug/
+--- wasm_modules: ngx_rust_tests
+--- tcp_listen: $TEST_NGINX_UNIX_SOCKET
+--- tcp_shutdown: 2
+--- tcp_reply eval
+sub {
+    return ["HTTP/1.1 100 OK\r\n",
+            "Connection: close\r\n",
+            "Content-Length: 0\r\n",
+            "\r\n"];
+}
+--- config
+    location /t {
+        wasm_call header_filter ngx_rust_tests log_notice_hello;
+        wasm_call body_filter ngx_rust_tests log_notice_hello;
+        proxy_pass http://unix:$TEST_NGINX_UNIX_SOCKET:/;
+    }
+--- error_code: 502
+--- ignore_response
+--- grep_error_log eval: qr/(upstream prematurely closed|(wasm ops calling .*? in .*? phase))/
+--- grep_error_log_out eval
+qr/\Aupstream prematurely closed
+wasm ops calling "ngx_rust_tests\.log_notice_hello" in "header_filter" phase
+wasm ops calling "ngx_rust_tests\.log_notice_hello" in "body_filter" phase\Z/
+--- no_error_log
+[crit]
+[emerg]
+[alert]
+
+
+
+=== TEST 2: upstream connection abort - chained filters
+Filters on aborted upstream connections execute response phases on the produced
+HTTP 502 response.
+--- wasm_modules: on_phases
+--- tcp_listen: $TEST_NGINX_UNIX_SOCKET
+--- tcp_shutdown: 2
+--- tcp_reply eval
+sub {
+    return ["HTTP/1.1 100 OK\r\n",
+            "Connection: close\r\n",
+            "Content-Length: 0\r\n",
+            "\r\n"];
+}
+--- config
+    location /t {
+        proxy_wasm on_phases;
+        proxy_wasm on_phases;
+        proxy_pass http://unix:$TEST_NGINX_UNIX_SOCKET:/;
+    }
+--- error_code: 502
+--- ignore_response
+--- grep_error_log eval: qr/\[(error|info)\] .*? ((upstream prematurely closed)|on_request_headers|on_response_headers|on_log|on_done|on_http_call_response.*)/
+--- grep_error_log_out eval
+qr/\A\[info\] .*? on_request_headers
+\[info\] .*? on_request_headers
+\[error\] .*? upstream prematurely closed
+\[info\] .*? on_response_headers
+\[info\] .*? on_response_headers
+\[info\] .*? on_done
+\[info\] .*? on_log
+\[info\] .*? on_done
+\[info\] .*? on_log\Z/
+--- no_error_log
+[crit]
+[emerg]
+[alert]

t/lib/proxy-wasm-tests/hostcalls/src/filter.rs

@@ -105,6 +105,12 @@ impl Context for TestHttp {
 
         self.resume_http_request()
     }
+
+    fn on_done(&mut self) -> bool {
+        info!("[hostcalls] on_done");
+
+        true
+    }
 }
 
 impl HttpContext for TestHttp {

t/lib/proxy-wasm-tests/on-phases/src/filter.rs

@@ -109,7 +109,23 @@ impl OnPhases {
     }
 }
 
-impl Context for OnPhases {}
+impl Context for OnPhases {
+    fn on_done(&mut self) -> bool {
+        info!("#{} on_done", self.context_id);
+
+        let log_msg = self
+            .config
+            .get("log_msg")
+            .map_or(String::new(), |s| s.to_string());
+
+        if !log_msg.is_empty() {
+            info!("#{} log_msg: {}", self.context_id, log_msg);
+        }
+
+        true
+    }
+}
+
 impl HttpContext for OnPhases {
     fn on_http_request_headers(&mut self, nheaders: usize, _eof: bool) -> Action {
         info!(