From 9cc7f1d75f84898d233957a94a560fc6839bb56a Mon Sep 17 00:00:00 2001 From: if0ne Date: Fri, 20 Feb 2026 15:45:40 +0300 Subject: [PATCH 1/7] feat: add on_push callback for handling data for sockets in pool Signed-off-by: if0ne --- src/ngx_http_lua_socket_tcp.c | 125 +++++++++++++++++++++++++++++++++- src/ngx_http_lua_socket_tcp.h | 3 + 2 files changed, 126 insertions(+), 2 deletions(-) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 93b0d119e1..4f9cf9fd19 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,20 @@ 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 +1234,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 +4543,22 @@ 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); @@ -5643,8 +5672,7 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) /* When the server closes the connection, * 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) + if (c == NULL || u->read_closed || u->write_closed) { lua_pushnil(L); lua_pushliteral(L, "closed"); @@ -5740,6 +5768,14 @@ 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 +5802,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"); @@ -5983,6 +6022,71 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) return NGX_OK; } + item = c->data; + + if (n > 0 && item->on_push_cb_ref != LUA_NOREF) { + char rbuf[4096]; + ssize_t nread; + lua_State *L; + int close_conn; + int nret; + + 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; + } + + rbuf[0] = buf[0]; + + /* read the available data into a stack buffer; no request pool needed */ + nread = c->recv(c, rbuf + 1, sizeof(rbuf) - 1); + if (nread <= 0) { + goto close; + } + + lua_rawgeti(L, LUA_REGISTRYINDEX, item->on_push_cb_ref); + lua_pushlstring(L, rbuf, (size_t) nread); + + /* callback(data) -> reply:string|nil, close: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, sdata + nsent, slen - nsent, 0); + if (w <= 0) { + lua_pop(L, 2); + goto close; + } + nsent += w; + } + } + + close_conn = !lua_toboolean(L, -1); + lua_pop(L, 2); + + if (!close_conn) { + if (ngx_handle_read_event(c->read, 0) != NGX_OK) { + goto close; + } + return NGX_OK; + } + } + close: ngx_log_debug1(NGX_LOG_DEBUG_HTTP, ev->log, 0, @@ -5991,6 +6095,14 @@ 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 +6154,15 @@ 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; From 9d018ac7d4c4c6b8337425a966c017925aa82528 Mon Sep 17 00:00:00 2001 From: if0ne Date: Wed, 11 Mar 2026 13:31:12 +0300 Subject: [PATCH 2/7] fix: comment line Signed-off-by: if0ne --- src/ngx_http_lua_socket_tcp.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 4f9cf9fd19..6ff46886a9 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -6042,7 +6042,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) rbuf[0] = buf[0]; - /* read the available data into a stack buffer; no request pool needed */ + /* read the available data into a stack buffer */ nread = c->recv(c, rbuf + 1, sizeof(rbuf) - 1); if (nread <= 0) { goto close; From 8093d8feeb3be7910dbb0e9d8ac63af67189c5de Mon Sep 17 00:00:00 2001 From: if0ne Date: Wed, 11 Mar 2026 13:44:33 +0300 Subject: [PATCH 3/7] fix: code style Signed-off-by: if0ne --- src/ngx_http_lua_socket_tcp.c | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 6ff46886a9..ae31cfed09 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -1087,6 +1087,7 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) 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", @@ -4555,6 +4556,7 @@ ngx_http_lua_socket_tcp_finalize(ngx_http_request_t *r, if (L != NULL) { luaL_unref(L, LUA_REGISTRYINDEX, u->on_push_cb_ref); } + u->on_push_cb_ref = LUA_NOREF; } @@ -5773,6 +5775,7 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) if (evict_L != NULL) { luaL_unref(evict_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); } + item->on_push_cb_ref = LUA_NOREF; } @@ -6072,6 +6075,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) lua_pop(L, 2); goto close; } + nsent += w; } } @@ -6100,6 +6104,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) if (close_L != NULL) { luaL_unref(close_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); } + item->on_push_cb_ref = LUA_NOREF; } @@ -6160,6 +6165,7 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) if (close_L != NULL) { luaL_unref(close_L, LUA_REGISTRYINDEX, item->on_push_cb_ref); } + item->on_push_cb_ref = LUA_NOREF; } From 49b83eef56402ea0b9c856a1dd98a603ca511337 Mon Sep 17 00:00:00 2001 From: if0ne Date: Wed, 11 Mar 2026 14:23:49 +0300 Subject: [PATCH 4/7] fix: warnings and errors Signed-off-by: if0ne --- src/ngx_http_lua_socket_tcp.c | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index ae31cfed09..7abd1d87be 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -6028,11 +6028,10 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) item = c->data; if (n > 0 && item->on_push_cb_ref != LUA_NOREF) { - char rbuf[4096]; - ssize_t nread; - lua_State *L; - int close_conn; - int nret; + unsigned char rbuf[4096]; + ssize_t nread; + lua_State *L; + int close_conn; ngx_log_debug0(NGX_LOG_DEBUG_HTTP, ev->log, 0, "lua tcp socket keepalive: data received, calling cb"); @@ -6052,7 +6051,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) } lua_rawgeti(L, LUA_REGISTRYINDEX, item->on_push_cb_ref); - lua_pushlstring(L, rbuf, (size_t) nread); + lua_pushlstring(L, (const char *) rbuf, (size_t) nread); /* callback(data) -> reply:string|nil, close:bool */ if (lua_pcall(L, 1, 2, 0) != LUA_OK) { @@ -6070,7 +6069,10 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) ssize_t nsent = 0; while ((size_t) nsent < slen) { - ssize_t w = c->send(c, sdata + nsent, slen - nsent, 0); + ssize_t w = c->send(c, + (u_char *) (sdata + nsent), + slen - nsent); + if (w <= 0) { lua_pop(L, 2); goto close; From 7f9c499139c2f105af1bfc6e057b3dce233a790c Mon Sep 17 00:00:00 2001 From: if0ne Date: Wed, 11 Mar 2026 16:40:28 +0300 Subject: [PATCH 5/7] fix: restore eof check Signed-off-by: if0ne --- src/ngx_http_lua_socket_tcp.c | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 7abd1d87be..61e590ad41 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -5674,7 +5674,9 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) /* When the server closes the connection, * epoll will return EPOLLRDHUP event and nginx will set pending_eof. */ - if (c == NULL || u->read_closed || u->write_closed) + if (c == NULL || u->read_closed || u->write_closed + || (u->on_push_cb_ref == LUA_NOREF + && (c->read->eof || c->read->pending_eof))) { lua_pushnil(L); lua_pushliteral(L, "closed"); From 3b2aded5df977352a45a08437aaacf2fad0213ee Mon Sep 17 00:00:00 2001 From: Dmitry Dorofeev Date: Sat, 19 Sep 2026 17:55:03 +0300 Subject: [PATCH 6/7] tests: exercise pooled socket push callbacks with a TCP fixture. --- .github/workflows/socket-on-push.yml | 80 ++++++ doc/socket-on-push-tests.md | 26 ++ t/190-socket-on-push.t | 416 +++++++++++++++++++++++++++ t/lib/socket-on-push.lua | 63 ++++ t/lib/tcp-push-server.py | 90 ++++++ 5 files changed, 675 insertions(+) create mode 100644 .github/workflows/socket-on-push.yml create mode 100644 doc/socket-on-push-tests.md create mode 100644 t/190-socket-on-push.t create mode 100644 t/lib/socket-on-push.lua create mode 100644 t/lib/tcp-push-server.py 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/t/190-socket-on-push.t b/t/190-socket-on-push.t new file mode 100644 index 0000000000..8a34565fc3 --- /dev/null +++ b/t/190-socket-on-push.t @@ -0,0 +1,416 @@ +# 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) { + kill 'TERM', $PushPid; + close $PushServer; + } +} + +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 + return "NEW\r\n", 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() From 25f5ca3924c7bdf6414b87f9b1e171b0b195bf16 Mon Sep 17 00:00:00 2001 From: Dmitry Dorofeev Date: Sat, 19 Sep 2026 18:00:09 +0300 Subject: [PATCH 7/7] fix: preserve pushed bytes and release callbacks on pool checkout. --- src/ngx_http_lua_socket_tcp.c | 36 ++++++++++++++++------------------- t/190-socket-on-push.t | 7 ++++++- 2 files changed, 22 insertions(+), 21 deletions(-) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 61e590ad41..f4221c17a0 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -5944,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; @@ -5993,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; @@ -6014,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 */ @@ -6027,11 +6037,7 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) return NGX_OK; } - item = c->data; - if (n > 0 && item->on_push_cb_ref != LUA_NOREF) { - unsigned char rbuf[4096]; - ssize_t nread; lua_State *L; int close_conn; @@ -6044,18 +6050,10 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) goto close; } - rbuf[0] = buf[0]; - - /* read the available data into a stack buffer */ - nread = c->recv(c, rbuf + 1, sizeof(rbuf) - 1); - if (nread <= 0) { - goto close; - } - lua_rawgeti(L, LUA_REGISTRYINDEX, item->on_push_cb_ref); - lua_pushlstring(L, (const char *) rbuf, (size_t) nread); + lua_pushlstring(L, (const char *) buf, (size_t) n); - /* callback(data) -> reply:string|nil, close:bool */ + /* 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", @@ -6088,10 +6086,8 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) lua_pop(L, 2); if (!close_conn) { - if (ngx_handle_read_event(c->read, 0) != NGX_OK) { - goto close; - } - return NGX_OK; + /* Drain all available data before rearming the read event. */ + goto again; } } diff --git a/t/190-socket-on-push.t b/t/190-socket-on-push.t index 8a34565fc3..69153e6a3b 100644 --- a/t/190-socket-on-push.t +++ b/t/190-socket-on-push.t @@ -18,8 +18,10 @@ our $LuaLib = abs_path('t/lib'); END { if ($PushPid) { + my $status = $?; kill 'TERM', $PushPid; close $PushServer; + $? = $status; } } @@ -346,7 +348,10 @@ callbacks: 0 local received = "" sock = t.connect("replace", function(data) received = received .. data - return "NEW\r\n", true + if received == "PUSH\r\n" then + return "NEW\r\n", true + end + return nil, true end) collectgarbage() collectgarbage()