Stream: do not use the filter output chain after the filter returns

Asynchronous s.send() without a direction appended to the output chain
of the stream filter call, which was a local variable.  The pointer to
it was kept in the context after the filter returned.  Only calls that
passed an options object were rejected, so a plain s.send(data) from a
timer or a promise wrote to a returned stack frame with both engines.

Reset the pointer when the filter returns and reject asynchronous sends
that have no direction regardless of the options argument.

Changed behaviour:
- Asynchronous s.send(data, options) without a direction in the options
  throws TypeError, as s.send(data) already did for the options-less
  call, instead of writing to a returned stack frame.

diff --git a/nginx/ngx_stream_js_module.c b/nginx/ngx_stream_js_module.c
index 334c04f8b5c867435849415d1844f2731b4ef8b1..aac1ebe537dd2c0ae01f0c6f649200d90d4c6a15 100644 (file)
--- a/nginx/ngx_stream_js_module.c
+++ b/nginx/ngx_stream_js_module.c
@@ -1068,12 +1068,16 @@ ngx_stream_js_body_filter(ngx_stream_session_t *s, ngx_chain_t *in,
     ctx->last_out = &out;
 
     rc = ctx->body_filter(s, ctx, in, from_upstream);
+
+    ctx->buf = NULL;
+
     if (rc != NGX_OK) {
+        ctx->last_out = NULL;
         return NGX_ERROR;
     }
 
-    ctx->buf = NULL;
     *ctx->last_out = NULL;
+    ctx->last_out = NULL;
 
     return ngx_stream_js_next_filter(s, ctx, out, from_upstream);
 }
@@ -1709,13 +1713,13 @@ ngx_stream_js_ext_send(njs_vm_t *vm, njs_value_t *args, njs_uint_t nargs,
             if (value != NULL) {
                 from_upstream = njs_value_bool(value);
             }
-
-            if (value == NULL && ctx->buf == NULL) {
-                goto exception;
-            }
         }
     }
 
+    if (from_upstream == NGX_JS_BOOL_UNSET && ctx->last_out == NULL) {
+        goto exception;
+    }
+
     cl = ngx_chain_get_free_buf(c->pool, &ctx->free);
     if (cl == NULL) {
         njs_vm_memory_error(vm);
@@ -2466,15 +2470,14 @@ ngx_stream_qjs_ext_send(JSContext *cx, JSValueConst this_val, int argc,
                 from_upstream = JS_ToBool(cx, val);
                 JS_FreeValue(cx, val);
             }
-
-            if (from_upstream == NGX_JS_BOOL_UNSET && ctx->buf == NULL) {
-                return JS_ThrowTypeError(cx, "from_upstream flag is "
-                                         "expected when called "
-                                         "asynchronously");
-            }
         }
     }
 
+    if (from_upstream == NGX_JS_BOOL_UNSET && ctx->last_out == NULL) {
+        return JS_ThrowTypeError(cx, "from_upstream flag is expected when "
+                                 "called asynchronously");
+    }
+
     val = argv[0];
 
     if (JS_IsNullOrUndefined(val)) {

diff --git a/nginx/t/stream_js_send_async.t b/nginx/t/stream_js_send_async.t
new file mode 100644 (file)
index 0000000..8c4c2ae
--- /dev/null
+++ b/nginx/t/stream_js_send_async.t
@@ -0,0 +1,139 @@
+#!/usr/bin/perl
+
+# (C) Dmitry Volyntsev
+# (C) F5, Inc.
+
+# Tests for asynchronous s.send() calls in the stream js filter.  Output
+# chain pointers belong to a filter call and must not be used after it.
+
+###############################################################################
+
+use warnings;
+use strict;
+
+use Test::More;
+
+BEGIN { use FindBin; chdir($FindBin::Bin); }
+
+use lib 'lib';
+use Test::Nginx;
+use Test::Nginx::Stream qw/ stream /;
+
+###############################################################################
+
+select STDERR; $| = 1;
+select STDOUT; $| = 1;
+
+my $t = Test::Nginx->new()->has(qw/stream/)
+    ->write_file_expand('nginx.conf', <<'EOF');
+
+%%TEST_GLOBALS%%
+
+daemon off;
+
+events {
+}
+
+stream {
+    %%TEST_GLOBALS_STREAM%%
+
+    js_import test.js;
+
+    server {
+        listen      127.0.0.1:8081;
+        js_filter   test.send;
+        proxy_pass  127.0.0.1:8090;
+    }
+
+    server {
+        listen      127.0.0.1:8082;
+        js_filter   test.send_options;
+        proxy_pass  127.0.0.1:8090;
+    }
+
+    server {
+        listen      127.0.0.1:8083;
+        js_filter   test.send_direct;
+        proxy_pass  127.0.0.1:8090;
+    }
+}
+
+EOF
+
+$t->write_file('test.js', <<'EOF');
+    function later(s, fn) {
+        s.on('upload', () => {
+            s.off('upload');
+
+            setTimeout(() => {
+                try {
+                    fn();
+
+                } catch (e) {
+                    s.sendDownstream(e instanceof TypeError ? 'TypeError'
+                                                            : e.name);
+                }
+            }, 1);
+        });
+    }
+
+    function send(s) {
+        later(s, () => {
+            s.send('late');
+            s.sendDownstream('accepted');
+        });
+    }
+
+    function send_options(s) {
+        later(s, () => {
+            s.send('late', {last: false});
+            s.sendDownstream('accepted');
+        });
+    }
+
+    function send_direct(s) {
+        later(s, () => {
+            s.sendDownstream('a');
+            s.send('b', {from_upstream: true});
+        });
+    }
+
+    export default {send, send_options, send_direct};
+EOF
+
+$t->run_daemon(\&stream_daemon, port(8090));
+$t->try_run('no stream njs available')->plan(3);
+$t->waitforsocket('127.0.0.1:' . port(8090));
+
+###############################################################################
+
+is(stream('127.0.0.1:' . port(8081))->io('go'), 'TypeError',
+    'async send without direction');
+is(stream('127.0.0.1:' . port(8082))->io('go'), 'TypeError',
+    'async send with options without direction');
+is(stream('127.0.0.1:' . port(8083))->io('go', read => 2), 'ab',
+    'async send with direction');
+
+$t->stop();
+
+###############################################################################
+
+sub stream_daemon {
+    my $server = IO::Socket::INET->new(
+        Proto => 'tcp',
+        LocalAddr => '127.0.0.1:' . port(8090),
+        Listen => 5,
+        Reuse => 1
+    )
+        or die "Can't create listening socket: $!\n";
+
+    local $SIG{PIPE} = 'IGNORE';
+
+    while (my $client = $server->accept()) {
+        $client->autoflush(1);
+        $client->sysread(my $buffer, 65536);
+        close $client;
+    }
+}
+
+###############################################################################