test(port): add IPC failure injection coverage

auto/sources

@@ -172,6 +172,7 @@ NXT_TEST_SRCS=" \
     src/test/nxt_http_parse_test.c \
     src/test/nxt_strverscmp_test.c \
     src/test/nxt_base64_test.c \
+    src/test/nxt_port_fail_test.c \
 "
 
 

src/nxt_cert.c

@@ -1241,6 +1241,7 @@ nxt_cert_store_get_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
          * dereference it through "%FN" on a close-failure log path.
          */
         nxt_fd_close(file.fd);
+        file.fd = -1;
     }
 }
 

src/nxt_mp.c

@@ -305,6 +305,17 @@ nxt_mp_release(nxt_mp_t *mp)
 }
 
 
+#if (NXT_TESTS)
+
+uint32_t
+nxt_mp_test_retain_count(nxt_mp_t *mp)
+{
+    return mp->retain;
+}
+
+#endif
+
+
 void
 nxt_mp_destroy(nxt_mp_t *mp)
 {

src/nxt_mp.h

@@ -52,6 +52,16 @@ NXT_EXPORT void nxt_mp_retain(nxt_mp_t *mp);
  */
 NXT_EXPORT void nxt_mp_release(nxt_mp_t *mp);
 
+#if (NXT_TESTS)
+/*
+ * Returns the current retention count.  Used by
+ * src/test/nxt_port_fail_test.c to verify the "retain after
+ * successful send" pattern leaves the pool's retain at its
+ * baseline (1) when the send fails.
+ */
+NXT_EXPORT uint32_t nxt_mp_test_retain_count(nxt_mp_t *mp);
+#endif
+
 /* nxt_mp_test_sizes() tests validity of memory pool parameters. */
 NXT_EXPORT nxt_bool_t nxt_mp_test_sizes(size_t cluster_size,
     size_t page_alignment, size_t page_size, size_t min_chunk_size);

src/nxt_port.h

@@ -349,6 +349,11 @@ nxt_int_t nxt_port_socket_write2(nxt_task_t *task, nxt_port_t *port,
     nxt_uint_t type, nxt_fd_t fd, nxt_fd_t fd2, uint32_t stream,
     nxt_port_id_t reply_port, nxt_buf_t *b);
 
+#if (NXT_TESTS)
+void nxt_port_test_msg_alloc_failures(nxt_uint_t failures);
+void nxt_port_test_run_error_handler(nxt_task_t *task, nxt_port_t *port);
+#endif
+
 nxt_inline nxt_int_t
 nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port,
     nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port,

src/nxt_port_rpc.c

@@ -29,6 +29,39 @@ static void
 nxt_port_rpc_remove_from_peers(nxt_task_t *task, nxt_port_t *port,
     nxt_port_rpc_reg_t *reg);
 
+#if (NXT_TESTS)
+static nxt_uint_t  nxt_port_rpc_test_alloc_failure_count;
+static nxt_uint_t  nxt_port_rpc_test_insert_failure_count;
+
+
+void
+nxt_port_rpc_test_alloc_failures(nxt_uint_t failures)
+{
+    nxt_port_rpc_test_alloc_failure_count = failures;
+}
+
+
+void
+nxt_port_rpc_test_insert_failures(nxt_uint_t failures)
+{
+    nxt_port_rpc_test_insert_failure_count = failures;
+}
+
+
+static nxt_bool_t
+nxt_port_rpc_test_should_fail(nxt_uint_t *failures)
+{
+    if (*failures == 0) {
+        return 0;
+    }
+
+    (*failures)--;
+
+    return 1;
+}
+
+#endif
+
 
 nxt_int_t
 nxt_port_rpc_init(void)
@@ -130,6 +163,16 @@ nxt_port_rpc_register_handler_ex(nxt_task_t *task, nxt_port_t *port,
 
     stream = nxt_atomic_fetch_add(nxt_stream_ident, 1);
 
+#if (NXT_TESTS)
+    if (nxt_slow_path(nxt_port_rpc_test_should_fail(
+                          &nxt_port_rpc_test_alloc_failure_count)))
+    {
+        nxt_debug(task, "rpc: stream #%uD failed to allocate reg", stream);
+
+        return NULL;
+    }
+#endif
+
     reg = nxt_mp_zalloc(port->mem_pool, sizeof(nxt_port_rpc_reg_t) + ex_size);
 
     if (nxt_slow_path(reg == NULL)) {
@@ -149,6 +192,18 @@ nxt_port_rpc_register_handler_ex(nxt_task_t *task, nxt_port_t *port,
     lhq.value = reg;
     lhq.pool = port->mem_pool;
 
+#if (NXT_TESTS)
+    if (nxt_slow_path(nxt_port_rpc_test_should_fail(
+                          &nxt_port_rpc_test_insert_failure_count)))
+    {
+        nxt_debug(task, "rpc: stream #%uD failed to add reg", stream);
+
+        nxt_mp_free(port->mem_pool, reg);
+
+        return NULL;
+    }
+#endif
+
     switch (nxt_lvlhsh_insert(&port->rpc_streams, &lhq)) {
 
     case NXT_OK:

src/nxt_port_rpc.h

@@ -20,6 +20,11 @@ void *nxt_port_rpc_register_handler_ex(nxt_task_t *task, nxt_port_t *port,
     nxt_port_rpc_handler_t ready_handler, nxt_port_rpc_handler_t error_handler,
     size_t ex_size);
 
+#if (NXT_TESTS)
+void nxt_port_rpc_test_alloc_failures(nxt_uint_t failures);
+void nxt_port_rpc_test_insert_failures(nxt_uint_t failures);
+#endif
+
 uint32_t nxt_port_rpc_ex_stream(void *ex);
 void nxt_port_rpc_ex_set_peer(nxt_task_t *task, nxt_port_t *port,
     void *ex, nxt_pid_t peer);
@@ -32,4 +37,3 @@ void nxt_port_rpc_close(nxt_task_t *task, nxt_port_t *port);
 
 
 #endif /* _NXT_PORT_RPC_H_INCLUDED_ */
-

src/nxt_port_socket.c

@@ -37,6 +37,35 @@ static nxt_buf_t *nxt_port_buf_alloc(nxt_port_t *port);
 static void nxt_port_buf_free(nxt_port_t *port, nxt_buf_t *b);
 static void nxt_port_error_handler(nxt_task_t *task, void *obj, void *data);
 
+#if (NXT_TESTS)
+static nxt_uint_t  nxt_port_test_msg_alloc_failure_count;
+
+
+void
+nxt_port_test_msg_alloc_failures(nxt_uint_t failures)
+{
+    nxt_port_test_msg_alloc_failure_count = failures;
+}
+
+
+/*
+ * Public wrapper that lets src/test/nxt_port_fail_test.c invoke the
+ * static nxt_port_error_handler() directly with a synthesised port
+ * and queued message — used to verify that the "queued, then write
+ * failed" cleanup matches the ordering the cert/script/socket reply
+ * paths now mirror after the audit fix (close fd first, queue buffer
+ * completion second).  Pass NULL for `data` so use_delta does not
+ * decrement for the "obj == data" case — the test owns the port
+ * reference and releases it explicitly.
+ */
+void
+nxt_port_test_run_error_handler(nxt_task_t *task, nxt_port_t *port)
+{
+    nxt_port_error_handler(task, &port->socket, NULL);
+}
+
+#endif
+
 
 nxt_int_t
 nxt_port_socket_init(nxt_task_t *task, nxt_port_t *port, size_t max_size)
@@ -336,6 +365,13 @@ nxt_port_msg_alloc(const nxt_port_send_msg_t *m)
 {
     nxt_port_send_msg_t  *msg;
 
+#if (NXT_TESTS)
+    if (nxt_slow_path(nxt_port_test_msg_alloc_failure_count != 0)) {
+        nxt_port_test_msg_alloc_failure_count--;
+        return NULL;
+    }
+#endif
+
     msg = nxt_malloc(sizeof(nxt_port_send_msg_t));
     if (nxt_slow_path(msg == NULL)) {
         return NULL;

src/nxt_script.c

@@ -606,6 +606,7 @@ nxt_script_store_get_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
          * log path.
          */
         nxt_fd_close(file.fd);
+        file.fd = -1;
     }
 }