Fall back to the system heap when a library realloc cannot be served

heap_realloc returned NULL when a block outgrew the largest size class or
the budget, and libuv aborts when growing its watcher array fails: a poll
on fd 8190 or above (about 4,000 connections) killed the process.
Move the block to the system heap instead, as malloc and calloc already
do. test_heap_guard_cpp polls fd 16380 with the heaps installed.

src/core/fixed_heap.hpp

@@ -111,6 +111,13 @@ public:
         deallocate_locked(p);
     }
 
+    // Bytes the caller may use in a block this heap returned.
+    std::size_t usable_size(void *p) noexcept
+    {
+        Guard guard(lock_);
+        return class_size(header_of(p)->cls) - kHeader;
+    }
+
     bool owns(const void *p) const noexcept
     {
         auto *b = static_cast<const std::byte *>(p);

src/core/heap_guard.cpp

@@ -3,6 +3,7 @@
 
 #include <atomic>
 #include <cstdlib>
+#include <cstring>
 #include <memory>
 
 #include <ares.h>
@@ -54,14 +55,25 @@ void *heap_realloc(FixedHeap &h, void *old, std::size_t size) noexcept
     if (old != nullptr && !h.owns(old)) {
         return std::realloc(old, size); // heap-ok: allocated before the switch
     }
+    if (old != nullptr && size == 0) {
+        h.deallocate(old);
+        return nullptr;
+    }
+    // Remember the old block's size first: reallocate() frees it only when
+    // it succeeds.
+    std::size_t usable = old != nullptr ? h.usable_size(old) : 0;
     if (void *p = h.reallocate(old, size)) {
         return p;
     }
-    if (size == 0) {
-        return nullptr;
-    }
+    // Too large for any size class, or over budget: move to the system heap
+    // (libuv aborts when growing its watcher array fails).
     fallbacks.fetch_add(1, std::memory_order_relaxed);
-    return nullptr; // the caller keeps `old`
+    void *fresh = std::malloc(size); // heap-ok: over budget, see heap_report()
+    if (fresh != nullptr && old != nullptr) {
+        std::memcpy(fresh, old, usable < size ? usable : size);
+        h.deallocate(old);
+    }
+    return fresh; // on failure the caller keeps `old`
 }
 
 void heap_free(FixedHeap &h, void *p) noexcept

tests/CMakeLists.txt

@@ -83,6 +83,12 @@ ss_add_test(test_event_cpp "test_event_cpp.cpp" "ss_uv")
 ss_strict_warnings(test_event_cpp.cpp)
 set_tests_properties(test_event_cpp PROPERTIES TIMEOUT 30)
 
+# test_heap_guard_cpp - library heaps fall back instead of failing a realloc
+ss_add_test(test_heap_guard_cpp
+    "test_heap_guard_cpp.cpp;${PROJECT_SOURCE_DIR}/src/core/heap_guard.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp"
+    "ss_uv;ss_cares;ss_mbedcrypto;ss_sodium")
+ss_strict_warnings(test_heap_guard_cpp.cpp)
+
 # C++ sources under src/ must not allocate on the heap (see the script).
 find_package(Python3 COMPONENTS Interpreter QUIET)
 if(Python3_Interpreter_FOUND)

tests/test_heap_guard_cpp.cpp

@@ -0,0 +1,102 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+// The library heaps (src/core/heap_guard.cpp): a request no fixed heap can
+// serve must fall back to the system allocator instead of failing. libuv
+// aborts when growing its watcher array fails, which a poll on a high fd
+// triggers (the array grows to next_pow2(fd) pointers).
+#include <cstdio>
+#include <cstring>
+
+#ifndef _WIN32
+#include <sys/resource.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <unistd.h>
+#endif
+
+#include <uv.h>
+
+#include "core/heap_guard.hpp"
+#include "test_helpers.hpp"
+
+extern "C" {
+int verbose = 0;
+}
+
+namespace {
+
+void test_realloc_past_largest_class()
+{
+#ifndef _WIN32
+    constexpr int kHighFd = 16380; // 16384 * 8 bytes > the largest size class
+    rlimit lim{};
+    SS_CHECK(getrlimit(RLIMIT_NOFILE, &lim) == 0);
+    if (lim.rlim_cur <= static_cast<rlim_t>(kHighFd)) {
+        lim.rlim_cur = kHighFd + 16;
+        if (lim.rlim_max != RLIM_INFINITY && lim.rlim_max < lim.rlim_cur) {
+            std::printf("skip: RLIMIT_NOFILE hard limit %llu too low\n",
+                        static_cast<unsigned long long>(lim.rlim_max));
+            return;
+        }
+        if (setrlimit(RLIMIT_NOFILE, &lim) != 0) {
+            std::printf("skip: cannot raise RLIMIT_NOFILE\n");
+            return;
+        }
+    }
+    uv_loop_t loop;
+    SS_CHECK(uv_loop_init(&loop) == 0);
+    // A low fd first, so the watcher array starts small inside the heap.
+    int low = socket(AF_INET, SOCK_DGRAM, 0);
+    SS_CHECK(low >= 0);
+    uv_poll_t low_poll;
+    SS_CHECK(uv_poll_init(&loop, &low_poll, low) == 0);
+    SS_CHECK(uv_poll_start(&low_poll, UV_READABLE, [](uv_poll_t *, int, int) {}) == 0);
+    SS_CHECK(uv_run(&loop, UV_RUN_NOWAIT) >= 0);
+    std::size_t before = ss::heap_report().fallbacks;
+
+    int fd = socket(AF_INET, SOCK_DGRAM, 0);
+    SS_CHECK(fd >= 0);
+    SS_CHECK(dup2(fd, kHighFd) == kHighFd);
+    close(fd);
+    sockaddr_in addr{};
+    addr.sin_family      = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    socklen_t len        = sizeof(addr);
+    SS_CHECK(bind(kHighFd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) == 0);
+    SS_CHECK(getsockname(kHighFd, reinterpret_cast<sockaddr *>(&addr), &len) == 0);
+    static int readable = 0;
+    uv_poll_t poll;
+    SS_CHECK(uv_poll_init(&loop, &poll, kHighFd) == 0);
+    SS_CHECK(uv_poll_start(&poll, UV_READABLE, [](uv_poll_t *h, int status, int) {
+        if (status == 0) {
+            ++readable;
+            uv_poll_stop(h);
+        }
+    }) == 0);
+    SS_CHECK(ss::heap_report().fallbacks > before); // the array moved out of the heap
+    // Both watchers still work after the move.
+    SS_CHECK(sendto(low, "x", 1, 0, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) == 1);
+    for (int i = 0; i < 200 && readable == 0; ++i) { // 2 s, not forever
+        SS_CHECK(uv_run(&loop, UV_RUN_NOWAIT) >= 0);
+        usleep(10000);
+    }
+    SS_CHECK(readable == 1);
+    uv_close(reinterpret_cast<uv_handle_t *>(&low_poll), nullptr);
+    close(low);
+    uv_close(reinterpret_cast<uv_handle_t *>(&poll), nullptr);
+    SS_CHECK(uv_run(&loop, UV_RUN_DEFAULT) == 0);
+    SS_CHECK(uv_loop_close(&loop) == 0);
+    close(kHighFd);
+#endif
+}
+
+} // namespace
+
+int main()
+{
+    ss::LibraryHeapBudgets budgets;
+    budgets.uv    = 64 << 10;
+    budgets.cares = 256 << 10;
+    SS_CHECK(ss::install_library_heaps(budgets));
+    SS_RUN(test_realloc_past_largest_class);
+    return 0;
+}