fix(*) resolve a potential memleak in Lua bridge resolver

An occasional memory leak of Proxy-Wasm dispatch with Lua bridge
resolver, spotted by CI:

https://github.com/Kong/ngx_wasm_module/actions/runs/9884094484/job/27299927250

    ==19824== 224 bytes in 1 blocks are definitely lost in loss record 13 of 33
    ==19824==    at 0x4848899: malloc (in /usr/libexec/valgrind/vgpreload_memcheck-amd64-linux.so)
    ==19824==    by 0x1E56AA: ngx_alloc (ngx_alloc.c:22)
    ==19824==    by 0x1CF926: ngx_resolver_alloc (ngx_resolver.c:4400)
    ==19824==    by 0x1CF926: ngx_resolver_calloc.isra.0 (ngx_resolver.c:4413)
    ==19824==    by 0x1D1EFD: ngx_resolve_start (ngx_resolver.c:661)
    ==19824==    by 0x392C80: ngx_wasm_socket_tcp_connect (ngx_wasm_socket_tcp.c:360)
    ==19824==    by 0x3BC984: ngx_http_proxy_wasm_dispatch_resume_handler (ngx_http_proxy_wasm_dispatch.c:745)
    ==19824==    by 0x3BD4C3: ngx_http_proxy_wasm_dispatch_handler (ngx_http_proxy_wasm_dispatch.c:477)
    ==19824==    by 0x1DFBE9: ngx_event_process_posted (ngx_event_posted.c:35)
    ==19824==    by 0x1DF0E6: ngx_process_events_and_timers (ngx_event.c:273)
    ==19824==    by 0x1EE697: ngx_single_process_cycle (ngx_process_cycle.c:323)
    ==19824==    by 0x1A9078: main (nginx.c:384)

This was caused by having 2 parallel dispatch calls using
`pwm_lua_resolver` DNS resolution. When the first call responds and
produces a downstream response, it causes pending calls to be cancelled.

However when a cancelled call had not yet finished the Lua-bridge DNS
resolution, its associated `rslv_ctx` was never freed.

This adds a "cancel" feature to Lua bridge threads, which is invoked by
TCP sockets when they are closed while having a pending Lua resolver
thread.

Alas no test since we would need to mock an entire DNS server.

src/common/lua/ngx_wasm_lua.c

@@ -587,6 +587,46 @@ ngx_wasm_lua_thread_run(ngx_wasm_lua_ctx_t *lctx)
 }
 
 
+void
+ngx_wasm_lua_thread_cancel(ngx_wasm_lua_ctx_t *lctx)
+{
+    ngx_wasm_subsys_env_t    *env = lctx->env;
+#if (NGX_WASM_HTTP)
+    ngx_http_wasm_req_ctx_t  *rctx;
+    ngx_http_request_t       *r;
+    ngx_http_lua_ctx_t       *ctx;
+#endif
+
+    dd("enter");
+
+    ngx_wa_assert(!lctx->entry);
+    ngx_wa_assert(!lctx->finished);
+
+    lctx->cancelled = 1;
+
+    if (lctx->error_handler) {
+        (void) lctx->error_handler(lctx);
+    }
+
+    switch (env->subsys->kind) {
+#if (NGX_WASM_HTTP)
+    case NGX_WASM_SUBSYS_HTTP:
+        rctx = env->ctx.rctx;
+        r = rctx->r;
+        ctx = lctx->ctx.rlctx;
+
+        ngx_http_lua_del_thread(r, lctx->L, ctx, lctx->co_ctx);
+        break;
+#endif
+    default:
+        ngx_wasm_bad_subsystem(env);
+        break;
+    }
+
+    dd("exit");
+}
+
+
 /**
  * Return values:
  * NGX_OK:     no lua thread to run

src/common/lua/ngx_wasm_lua.h

@@ -50,6 +50,7 @@ struct ngx_wasm_lua_ctx_s {
     unsigned                        entry:1;        /* is entry lctx */
     unsigned                        yielded:1;      /* has yielded at least once */
     unsigned                        finished:1;     /* has finished */
+    unsigned                        cancelled:1;    /* cancelled */
 };
 
 
@@ -58,6 +59,7 @@ ngx_wasm_lua_ctx_t *ngx_wasm_lua_thread_new(const char *tag, const char *src,
     ngx_wasm_lua_handler_pt success_handler,
     ngx_wasm_lua_handler_pt error_handler);
 ngx_int_t ngx_wasm_lua_thread_run(ngx_wasm_lua_ctx_t *lctx);
+void ngx_wasm_lua_thread_cancel(ngx_wasm_lua_ctx_t *lctx);
 ngx_int_t ngx_wasm_lua_resume(ngx_wasm_subsys_env_t *env);
 
 

src/common/lua/ngx_wasm_lua_resolver.c

@@ -101,10 +101,15 @@ ngx_wasm_lua_resolver_error_handler(ngx_wasm_lua_ctx_t *lctx)
 
     dd("enter");
 
-    ngx_wa_assert(lctx->co_ctx->co_status == NGX_HTTP_LUA_CO_DEAD);
+#if (DDEBUG)
+    if (!lctx->cancelled) {
+        ngx_wa_assert(lctx->co_ctx->co_status == NGX_HTTP_LUA_CO_DEAD);
+    }
+
     ngx_wa_assert(!rslv_ctx->naddrs);
+#endif
 
-    if (lctx->yielded) {
+    if (lctx->yielded || lctx->cancelled) {
         /* re-enter normal resolver path (handler) if we yielded */
         rslv_ctx->state = NGX_WASM_LUA_RESOLVE_ERR;
 
@@ -165,6 +170,8 @@ ngx_wasm_lua_resolver_resolve(ngx_resolver_ctx_t *rslv_ctx)
         return NGX_ERROR;
     }
 
+    sock->lctx = lctx;
+
     /* args */
 
     host = ngx_palloc(lctx->pool, rslv_ctx->name.len + 1);

src/common/ngx_wasm_socket_tcp.c

@@ -349,12 +349,18 @@ ngx_wasm_socket_tcp_connect(ngx_wasm_socket_tcp_t *sock)
         if (resolver == NULL || !resolver->connections.nelts) {
             /* fallback to default resolver */
             wcf = ngx_wasm_core_cycle_get_conf(ngx_cycle);
+            resolver = wcf->resolver;
 
-            ngx_log_debug0(NGX_LOG_DEBUG_WASM, sock->log, 0,
-                           "wasm tcp socket using default resolver");
+            if (!rctx->pwm_lua_resolver) {
+                ngx_log_debug0(NGX_LOG_DEBUG_WASM, sock->log, 0,
+                               "wasm tcp socket using default resolver");
 
-            resolver = wcf->resolver;
-            rslv_tmp.timeout = wcf->resolver_timeout;
+                rslv_tmp.timeout = wcf->resolver_timeout;
+
+            } else {
+                ngx_log_debug0(NGX_LOG_DEBUG_WASM, sock->log, 0,
+                               "wasm tcp socket using lua resolver");
+            }
         }
 
         rslv_ctx = ngx_resolve_start(resolver, &rslv_tmp);
@@ -521,7 +527,11 @@ ngx_wasm_socket_resolve_handler(ngx_resolver_ctx_t *ctx)
 
     ngx_resolve_name_done(ctx);  /* frees ctx */
 
-    (void) ngx_wasm_socket_tcp_resume(sock);
+    if (!sock->closed) {
+        /* e.g. pwm_lua_resolver call was cancelled and socket closed, do not
+         * resume */
+        (void) ngx_wasm_socket_tcp_resume(sock);
+    }
 
     if (resume) {
         /* resolver error, continue request */
@@ -1230,6 +1240,13 @@ ngx_wasm_socket_tcp_destroy(ngx_wasm_socket_tcp_t *sock)
 
     ngx_wasm_socket_tcp_close(sock);
 
+#if (NGX_WASM_LUA)
+    if (sock->lctx && !sock->lctx->finished) {
+        /* cancel the pending lua resolver thread */
+        ngx_wasm_lua_thread_cancel(sock->lctx);
+    }
+#endif
+
     if (sock->host.data) {
         ngx_pfree(sock->pool, sock->host.data);
         sock->host.data = NULL;

src/common/ngx_wasm_socket_tcp.h

@@ -40,6 +40,9 @@ struct ngx_wasm_socket_tcp_s {
 
     ngx_wasm_socket_tcp_resume_handler_pt    resume_handler;
     void                                    *data;
+#if (NGX_WASM_LUA)
+    ngx_wasm_lua_ctx_t                      *lctx;
+#endif
 
     ngx_str_t                                host;
     ngx_wasm_upstream_resolved_t             resolved;