diff --git a/.github/workflows/socket-on-push.yml b/.github/workflows/socket-on-push.yml new file mode 100644 index 0000000000..d4237f3d46 --- /dev/null +++ b/.github/workflows/socket-on-push.yml @@ -0,0 +1,80 @@ +name: Socket on_push tests + +on: + push: + branches: [test/on-push-tcp, feat/callback-idle-conn] + pull_request: + paths: + - 'src/ngx_http_lua_socket_tcp.*' + - 't/190-socket-on-push.t' + - 't/lib/socket-on-push.lua' + - 't/lib/tcp-push-server.py' + - '.github/workflows/socket-on-push.yml' + +permissions: + contents: read + +jobs: + test: + name: nginx ${{ matrix.nginx }} / ${{ matrix.os }} + runs-on: ${{ matrix.os }} + timeout-minutes: 20 + strategy: + fail-fast: false + matrix: + nginx: ['1.27.1', '1.29.4'] + os: [ubuntu-22.04, ubuntu-24.04] + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 + + - name: Install build and test dependencies + run: | + sudo apt-get update -q + sudo apt-get install -y build-essential libpcre2-dev zlib1g-dev \ + libssl-dev cpanminus python3 + sudo cpanm --notest --mirror https://cpan.metacpan.org --mirror-only \ + Test::Nginx IPC::Run + + - name: Download sources + run: | + curl --fail --location --retry 3 \ + https://openresty.org/download/openresty-1.27.1.2.tar.gz | tar xz + curl --fail --location --retry 3 \ + https://nginx.org/download/nginx-${{ matrix.nginx }}.tar.gz | tar xz + git clone --branch v0.1.32 --depth=1 \ + https://github.com/openresty/lua-resty-core.git lua-resty-core + + - name: Build LuaJIT and nginx with the module under test + run: | + bundle="$PWD/openresty-1.27.1.2/bundle" + prefix="$RUNNER_TEMP/on-push" + make -C "$bundle/LuaJIT-2.1-20250117" -j2 PREFIX="$prefix/luajit" + make -C "$bundle/LuaJIT-2.1-20250117" install PREFIX="$prefix/luajit" + export LUAJIT_LIB="$prefix/luajit/lib" + export LUAJIT_INC="$prefix/luajit/include/luajit-2.1" + cd nginx-${{ matrix.nginx }} + ./configure --prefix="$prefix/nginx" --with-debug \ + --with-http_ssl_module --with-http_v2_module \ + --with-ld-opt="-Wl,-rpath,$LUAJIT_LIB" \ + --add-module="$bundle/ngx_devel_kit-0.3.3" \ + --add-module="$bundle/echo-nginx-module-0.63" \ + --add-module="$GITHUB_WORKSPACE" + make -j2 + make install + echo "$prefix/nginx/sbin" >> "$GITHUB_PATH" + echo "LUA_PATH=$GITHUB_WORKSPACE/lua-resty-core/lib/?.lua;$bundle/lua-resty-lrucache-0.15/lib/?.lua;;" >> "$GITHUB_ENV" + + - name: Run socket push integration and basic Lua regression tests + env: + TEST_NGINX_TIMEOUT: '10' + TEST_NGINX_SLEEP: '0.01' + run: | + nginx -V + prove -v t/190-socket-on-push.t t/000-sanity.t t/002-content.t + + - name: Upload nginx logs on failure + if: failure() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: nginx-logs-${{ matrix.nginx }}-${{ matrix.os }} + path: t/servroot/logs/ diff --git a/doc/socket-on-push-tests.md b/doc/socket-on-push-tests.md new file mode 100644 index 0000000000..fa0ce69365 --- /dev/null +++ b/doc/socket-on-push-tests.md @@ -0,0 +1,26 @@ +# Socket push integration tests + +Run from the repository root with a debug build of nginx containing this +checkout's ngx_lua module, matching lua-resty-core/lua-resty-lrucache on +`LUA_PATH`, the Perl `Test::Nginx` and `IPC::Run` modules, and Python 3: + +```sh +TEST_NGINX_BINARY=/path/to/nginx prove -v t/190-socket-on-push.t +``` + +The test starts and stops its own Python standard-library TCP server on an +OS-assigned loopback port. No NATS server or other service is needed. Each +test parks a real Lua cosocket in nginx's keepalive pool, then uses a separate +control connection to trigger server pushes. Callback replies are checked by +the server. Reuse checks verify both the reuse counter and bidirectional IO +on the original connection. + +Callbacks receive arbitrary chunks of a TCP byte stream, not application +messages. Tests explicitly cover a one-byte push, separated fragments, binary +data, and a payload larger than the read buffer, as well as reply/keep-open, +reply/close, callback failure, EOF, and the behavior without a callback. They +also check callback lifetime across request completion and pool checkout. + +The `Socket on_push tests` GitHub Actions workflow builds the module with +nginx 1.27.1 and 1.29.4 on Ubuntu 22.04 and 24.04, runs each push case twice, +and runs the existing basic Lua sanity/content tests. diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 93b0d119e1..f4221c17a0 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -576,6 +576,7 @@ ngx_http_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_http_request_t *r, for (i = 0; i < pool_size; i++) { ngx_queue_insert_head(&sp->free, &items[i].queue); items[i].socket_pool = sp; + items[i].on_push_cb_ref = LUA_NOREF; } *spool = sp; @@ -1003,6 +1004,7 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) int key_index; ngx_int_t backlog; ngx_int_t pool_size; + int on_push_cb_ref; ngx_str_t key; const char *msg; @@ -1036,6 +1038,7 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) key_index = 2; pool_size = 0; custom_pool = 0; + on_push_cb_ref = LUA_NOREF; llcf = ngx_http_get_module_loc_conf(r, ngx_http_lua_module); if (lua_type(L, n) == LUA_TTABLE) { @@ -1080,6 +1083,21 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) lua_pop(L, 1); + lua_getfield(L, n, "on_push"); + + if (lua_isfunction(L, -1)) { + on_push_cb_ref = luaL_ref(L, LUA_REGISTRYINDEX); + + } else { + if (!lua_isnil(L, -1)) { + msg = lua_pushfstring(L, "bad \"on_push\" option type: %s", + lua_typename(L, lua_type(L, -1))); + return luaL_argerror(L, n, msg); + } + + lua_pop(L, 1); + } + lua_getfield(L, n, "pool"); switch (lua_type(L, -1)) { @@ -1217,6 +1235,8 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) ngx_memzero(u, sizeof(ngx_http_lua_socket_tcp_upstream_t)); + u->on_push_cb_ref = on_push_cb_ref; + u->request = r; /* set the controlling request */ u->conf = llcf; @@ -4524,12 +4544,23 @@ ngx_http_lua_socket_tcp_finalize(ngx_http_request_t *r, { ngx_connection_t *c; ngx_http_lua_socket_pool_t *spool; + lua_State *L; dd("request: %p, u: %p, u->cleanup: %p", r, u, u->cleanup); ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, "lua finalize socket"); + if (u->on_push_cb_ref != LUA_NOREF) { + L = ngx_http_lua_get_lua_vm(r, NULL); + if (L != NULL) { + luaL_unref(L, LUA_REGISTRYINDEX, u->on_push_cb_ref); + } + + u->on_push_cb_ref = LUA_NOREF; + } + + if (u->cleanup) { *u->cleanup = NULL; ngx_http_lua_cleanup_free(r, u->cleanup); @@ -5644,7 +5675,8 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) * epoll will return EPOLLRDHUP event and nginx will set pending_eof. */ if (c == NULL || u->read_closed || u->write_closed - || c->read->eof || c->read->pending_eof) + || (u->on_push_cb_ref == LUA_NOREF + && (c->read->eof || c->read->pending_eof))) { lua_pushnil(L); lua_pushliteral(L, "closed"); @@ -5740,6 +5772,15 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) item = ngx_queue_data(q, ngx_http_lua_socket_pool_item_t, queue); + if (item->on_push_cb_ref != LUA_NOREF) { + lua_State *evict_L = spool->lua_vm; + if (evict_L != NULL) { + luaL_unref(evict_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); + } + + item->on_push_cb_ref = LUA_NOREF; + } + ngx_http_lua_socket_tcp_close_connection(item->connection); /* only decrease the counter for connections which were counted */ @@ -5766,6 +5807,9 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) item->connection = c; ngx_queue_insert_head(&spool->cache, q); + item->on_push_cb_ref = u->on_push_cb_ref; + u->on_push_cb_ref = LUA_NOREF; + ngx_log_debug0(NGX_LOG_DEBUG_HTTP, pc->log, 0, "lua tcp socket clear current socket connection"); @@ -5900,6 +5944,11 @@ ngx_http_lua_get_keepalive_peer(ngx_http_request_t *r, u->udata_queue = item->udata_queue; item->udata_queue = NULL; + if (item->on_push_cb_ref != LUA_NOREF) { + luaL_unref(spool->lua_vm, LUA_REGISTRYINDEX, item->on_push_cb_ref); + item->on_push_cb_ref = LUA_NOREF; + } + #if 1 u->write_event_handler = ngx_http_lua_socket_dummy_handler; u->read_event_handler = ngx_http_lua_socket_dummy_handler; @@ -5949,7 +5998,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) ngx_http_lua_socket_pool_t *spool; int n; - unsigned char buf[1]; + unsigned char buf[4096]; ngx_connection_t *c; c = ev->data; @@ -5970,8 +6019,13 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) ngx_log_debug0(NGX_LOG_DEBUG_HTTP, ev->log, 0, "lua tcp socket keepalive close handler check stale events"); + item = c->data; + +again: + /* consume the possible ssl-layer data implicitly */ - n = c->recv(c, buf, 1); + n = c->recv(c, buf, + item->on_push_cb_ref == LUA_NOREF ? 1 : sizeof(buf)); if (n == NGX_AGAIN) { /* stale event */ @@ -5983,6 +6037,60 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) return NGX_OK; } + if (n > 0 && item->on_push_cb_ref != LUA_NOREF) { + lua_State *L; + int close_conn; + + ngx_log_debug0(NGX_LOG_DEBUG_HTTP, ev->log, 0, + "lua tcp socket keepalive: data received, calling cb"); + + spool = item->socket_pool; + L = spool->lua_vm; + if (L == NULL) { + goto close; + } + + lua_rawgeti(L, LUA_REGISTRYINDEX, item->on_push_cb_ref); + lua_pushlstring(L, (const char *) buf, (size_t) n); + + /* callback(data) -> reply:string|nil, keepalive:bool */ + if (lua_pcall(L, 1, 2, 0) != LUA_OK) { + ngx_log_error(NGX_LOG_ERR, ev->log, 0, + "lua tcp socket keepalive callback error: %s", + lua_tostring(L, -1)); + lua_pop(L, 1); + goto close; + } + + /* stack: reply(-2), close(-1) */ + if (lua_type(L, -2) == LUA_TSTRING) { + size_t slen; + const char *sdata = lua_tolstring(L, -2, &slen); + ssize_t nsent = 0; + + while ((size_t) nsent < slen) { + ssize_t w = c->send(c, + (u_char *) (sdata + nsent), + slen - nsent); + + if (w <= 0) { + lua_pop(L, 2); + goto close; + } + + nsent += w; + } + } + + close_conn = !lua_toboolean(L, -1); + lua_pop(L, 2); + + if (!close_conn) { + /* Drain all available data before rearming the read event. */ + goto again; + } + } + close: ngx_log_debug1(NGX_LOG_DEBUG_HTTP, ev->log, 0, @@ -5991,6 +6099,15 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) item = c->data; spool = item->socket_pool; + if (item->on_push_cb_ref != LUA_NOREF) { + lua_State *close_L = spool->lua_vm; + if (close_L != NULL) { + luaL_unref(close_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); + } + + item->on_push_cb_ref = LUA_NOREF; + } + ngx_http_lua_socket_tcp_close_connection(c); ngx_queue_remove(&item->queue); @@ -6042,6 +6159,16 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) q = ngx_queue_head(&spool->cache); item = ngx_queue_data(q, ngx_http_lua_socket_pool_item_t, queue); + + if (item->on_push_cb_ref != LUA_NOREF) { + lua_State *close_L = spool->lua_vm; + if (close_L != NULL) { + luaL_unref(close_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); + } + + item->on_push_cb_ref = LUA_NOREF; + } + c = item->connection; ngx_http_lua_socket_tcp_close_connection(c); diff --git a/src/ngx_http_lua_socket_tcp.h b/src/ngx_http_lua_socket_tcp.h index 6cc6fc5876..8abd0ee5e6 100644 --- a/src/ngx_http_lua_socket_tcp.h +++ b/src/ngx_http_lua_socket_tcp.h @@ -123,6 +123,8 @@ struct ngx_http_lua_socket_tcp_upstream_s { ngx_http_lua_co_ctx_t *read_co_ctx; ngx_http_lua_co_ctx_t *write_co_ctx; + int on_push_cb_ref; + ngx_uint_t reused; struct sockaddr_storage sockaddr; socklen_t socklen; @@ -189,6 +191,7 @@ typedef struct { char host[COSOCKET_HOST_LEN]; ngx_uint_t reused; + int on_push_cb_ref; ngx_http_lua_socket_udata_queue_t *udata_queue; } ngx_http_lua_socket_pool_item_t; diff --git a/t/190-socket-on-push.t b/t/190-socket-on-push.t new file mode 100644 index 0000000000..69153e6a3b --- /dev/null +++ b/t/190-socket-on-push.t @@ -0,0 +1,421 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua; +use Cwd qw(abs_path); + +# Start a private TCP fixture on an OS-assigned loopback port. A separate +# control connection triggers pushes only after setkeepalive() has returned. +# No NATS installation, fixed ports, or timing-based server sleeps are needed. +our ($PushServer, $PushPid); +$PushPid = open $PushServer, '-|', 'python3', 't/lib/tcp-push-server.py' + or die "cannot start TCP push server: $!"; +my $port = <$PushServer>; +defined $port && $port =~ /^\d+\s*$/ + or die "TCP push server did not report its port"; +chomp $port; +$ENV{TEST_NGINX_PUSH_PORT} = $port; +our $LuaLib = abs_path('t/lib'); + +END { + if ($PushPid) { + my $status = $?; + kill 'TERM', $PushPid; + close $PushServer; + $? = $status; + } +} + +add_block_preprocessor(sub { + my $block = shift; + $block->set_value('main_config', 'env TEST_NGINX_PUSH_PORT;'); + $block->set_value('http_config', "lua_package_path '$LuaLib/?.lua;;';"); +}); + +repeat_each(2); +no_long_string(); +log_level('debug'); +plan tests => repeat_each() * (blocks() * 5); +run_tests(); + +__DATA__ + +=== TEST 1: reply to an idle push and reuse the same connection +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local received = "" + local sock = t.connect("reply", function(data) + received = received .. data + if received == "PING\r\n" then + return "PONG\r\n", true + end + return nil, true + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "reply", "PING\r\n") + t.command(control, "EXPECT", "reply", "PONG\r\n") + ngx.say("complete PING: ", received == "PING\r\n") + assert(t.reuse("reply", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +complete PING: true +reused: 1 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 2: nil reply keeps the connection open and preserves binary data +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local expected = "hello\0world\255\r\n" + local received = "" + local sock = t.connect("binary", function(data) + received = received .. data + return nil, true + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "binary", expected) + t.wait(function() return #received >= #expected end) + ngx.say("exact bytes: ", received == expected) + assert(t.reuse("binary", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +exact bytes: true +reused: 1 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 3: a single byte invokes the callback without closing the socket +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local received = "" + local sock = t.connect("single", function(data) + received = received .. data + return nil, true + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "single", "X") + t.wait(function() return #received > 0 end) + ngx.say("received: ", received) + assert(t.reuse("single", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +received: X +reused: 1 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 4: a callback can assemble a message from separate read events +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local received = "" + local calls = 0 + local sock = t.connect("fragmented", function(data) + calls = calls + 1 + received = received .. data + if received == "PING\r\n" then + return "PONG\r\n", true + end + return nil, true + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "fragmented", "P") + t.wait(function() return received == "P" end) + t.command(control, "SEND", "fragmented", "ING\r\n") + t.command(control, "EXPECT", "fragmented", "PONG\r\n") + ngx.say("multiple callbacks: ", calls >= 2) + assert(t.reuse("fragmented", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +multiple callbacks: true +reused: 1 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 5: drain a push larger than the callback buffer without losing bytes +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local expected = string.rep("0123456789\0", 1000) .. "END" + local received = "" + local sock = t.connect("large", function(data) + received = received .. data + return nil, true + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "large", expected) + t.wait(function() return #received >= #expected end) + ngx.say("exact bytes: ", received == expected) + assert(t.reuse("large", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +exact bytes: true +reused: 1 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 6: false sends the reply and closes the pooled connection +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local sock = t.connect("close", function(data) + return "BYE\r\n", false + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "close", "STOP\r\n") + t.command(control, "EXPECT", "close", "BYE\r\n") + t.command(control, "CLOSED", "close") + sock = t.connect("close") + ngx.say("reused: ", sock:getreusedtimes()) + assert(sock:close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +reused: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 7: a callback error closes the connection and leaves the worker usable +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local sock = t.connect("error", function(data) + error("deliberate on_push failure") + end) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "error", "PUSH\r\n") + t.command(control, "CLOSED", "error") + sock = t.connect("error") + ngx.say("reused: ", sock:getreusedtimes()) + assert(sock:close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +reused: 0 +--- error_log +lua tcp socket keepalive callback error: +--- no_error_log +[alert] +[crit] + + + +=== TEST 8: without a callback unsolicited data still closes the connection +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local sock = t.connect("default") + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "default", "PUSH\r\n") + t.command(control, "CLOSED", "default") + sock = t.connect("default") + ngx.say("reused: ", sock:getreusedtimes()) + assert(sock:close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +reused: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 9: EOF closes the pool item even with a callback +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local calls = 0 + local callbacks = setmetatable({}, { __mode = "v" }) + callbacks[1] = function(data) + calls = calls + 1 + return nil, true + end + local sock = t.connect("eof", callbacks[1]) + assert(sock:setkeepalive(10000)) + t.command(control, "CLOSE", "eof") + -- Collection proves nginx observed EOF and released the callback. + t.wait(function() + collectgarbage() + return callbacks[1] == nil + end) + sock = t.connect("eof") + ngx.say("reused: ", sock:getreusedtimes()) + ngx.say("callbacks: ", calls) + assert(sock:close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +reused: 0 +callbacks: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 10: checkout releases the old callback and re-pooling uses the new one +--- config + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + local callbacks = setmetatable({}, { __mode = "v" }) + callbacks[1] = function(data) return "OLD\r\n", true end + local sock = t.connect("replace", callbacks[1]) + assert(sock:setkeepalive(10000)) + collectgarbage() + ngx.say("pooled callback retained: ", callbacks[1] ~= nil) + + local received = "" + sock = t.connect("replace", function(data) + received = received .. data + if received == "PUSH\r\n" then + return "NEW\r\n", true + end + return nil, true + end) + collectgarbage() + collectgarbage() + ngx.say("old callback released: ", callbacks[1] == nil) + assert(sock:setkeepalive(10000)) + t.command(control, "SEND", "replace", "PUSH\r\n") + t.command(control, "EXPECT", "replace", "NEW\r\n") + ngx.say("new callback received: ", received == "PUSH\r\n") + assert(t.reuse("replace", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +pooled callback retained: true +old callback released: true +new callback received: true +reused: 2 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 11: callback survives completion of the request that pooled the socket +--- config + location /park { + content_by_lua_block { + local t = require "socket-on-push" + local state = { received = "" } + t.finished_request = state + local sock = t.connect("finished", function(data) + state.received = state.received .. data + if state.received == "PING\r\n" then + return "PONG\r\n", true + end + return nil, true + end) + assert(sock:setkeepalive(10000)) + } + } + location /t { + content_by_lua_block { + local t = require "socket-on-push" + local control = t.control() + assert(ngx.location.capture("/park").status == 200) + collectgarbage() + collectgarbage() + t.command(control, "SEND", "finished", "PING\r\n") + t.command(control, "EXPECT", "finished", "PONG\r\n") + ngx.say("callback survived: ", + t.finished_request.received == "PING\r\n") + assert(t.reuse("finished", control):close()) + assert(control:close()) + } + } +--- request +GET /t +--- response_body +callback survived: true +reused: 1 +--- no_error_log +[error] +[alert] +[crit] diff --git a/t/lib/socket-on-push.lua b/t/lib/socket-on-push.lua new file mode 100644 index 0000000000..8024d0310c --- /dev/null +++ b/t/lib/socket-on-push.lua @@ -0,0 +1,63 @@ +local _M = {} +local port = tonumber(os.getenv("TEST_NGINX_PUSH_PORT")) + +local function socket() + local sock = ngx.socket.tcp() + sock:settimeout(4000) + return sock +end + +function _M.control() + local sock = socket() + assert(sock:connect("127.0.0.1", port, { pool = "push-control" })) + assert(sock:send("CONTROL\n")) + return sock +end + +function _M.command(control, command, name, data) + if data then + local hex = data:gsub(".", function(c) + return string.format("%02x", string.byte(c)) + end) + command = command .. " " .. name .. " " .. hex + else + command = command .. " " .. name + end + assert(control:send(command .. "\n")) + local reply = assert(control:receive()) + assert(reply == "OK", reply) +end + +function _M.connect(name, callback) + local sock = socket() + assert(sock:connect("127.0.0.1", port, { + pool = name, pool_size = 1, on_push = callback, + })) + if sock:getreusedtimes() == 0 then + assert(sock:send("OPEN " .. name .. "\n")) + assert(sock:receive() == "READY") + end + return sock +end + +function _M.wait(predicate) + local deadline = ngx.now() + 3 + while not predicate() do + assert(ngx.now() < deadline, "push callback timed out") + ngx.sleep(0.001) + end +end + +function _M.reuse(name, control, callback) + local sock = _M.connect(name, callback) + ngx.say("reused: ", sock:getreusedtimes()) + -- Verify that it is still the server's original connection, in both + -- directions, rather than merely trusting the pool's reuse counter. + assert(sock:send("CHECK\r\n")) + _M.command(control, "EXPECT", name, "CHECK\r\n") + _M.command(control, "SEND", name, "ALIVE\r\n") + assert(sock:receive() == "ALIVE") + return sock +end + +return _M diff --git a/t/lib/tcp-push-server.py b/t/lib/tcp-push-server.py new file mode 100644 index 0000000000..bc2c7d3c1e --- /dev/null +++ b/t/lib/tcp-push-server.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +"""Loopback TCP fixture for socket-on-push.t (Python standard library only).""" + +import socket +import socketserver +import threading + + +class Peer: + def __init__(self, sock): + self.sock = sock + self.received = b"" + self.closed = False + self.changed = threading.Condition() + + def wait(self, predicate): + with self.changed: + if not self.changed.wait_for(predicate, timeout=3): + raise TimeoutError("peer did not reach the expected state") + + +peers = {} +peers_lock = threading.Lock() + + +class Handler(socketserver.StreamRequestHandler): + def handle(self): + self.request.settimeout(5) + self.request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + peer = None + try: + command = self.rfile.readline().decode().strip().split() + if command[0] == "OPEN": + peer = Peer(self.request) + with peers_lock: + peers[command[1]] = peer + self.wfile.write(b"READY\n") + while True: + data = self.request.recv(65536) + if not data: + break + with peer.changed: + peer.received += data + peer.changed.notify_all() + return + + if command != ["CONTROL"]: + raise ValueError("expected OPEN or CONTROL") + for line in self.rfile: + command = line.decode().strip().split() + with peers_lock: + target = peers[command[1]] + if command[0] == "SEND": + target.sock.sendall(bytes.fromhex(command[2])) + elif command[0] == "EXPECT": + expected = bytes.fromhex(command[2]) + target.wait(lambda: len(target.received) >= len(expected) + or target.closed) + with target.changed: + actual = target.received[:len(expected)] + if actual != expected: + raise ValueError("reply mismatch: " + repr(actual)) + target.received = target.received[len(expected):] + elif command[0] == "CLOSED": + target.wait(lambda: target.closed) + elif command[0] == "CLOSE": + target.sock.shutdown(socket.SHUT_RDWR) + else: + raise ValueError("unknown command") + self.wfile.write(b"OK\n") + except (OSError, ValueError, KeyError, IndexError) as exc: + try: + self.wfile.write(("ERROR " + str(exc) + "\n").encode()) + except OSError: + pass + finally: + if peer is not None: + with peer.changed: + peer.closed = True + peer.changed.notify_all() + + +class Server(socketserver.ThreadingTCPServer): + daemon_threads = True + + +if __name__ == "__main__": + with Server(("127.0.0.1", 0), Handler) as server: + print(server.server_address[1], flush=True) + server.serve_forever()