feature: callback on async push messages recieved on socket in the pool

.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/

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.

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);

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;