CCBC-1685: add lcb_trim_memory() to release cached pool memory

libcouchbase's per-pipeline netbuf allocator keeps released blocks on
an available list so the fast path can re-reserve without going back
to malloc.  A long-lived instance that occasionally bursts and then
idles retains the peak working set of every pipeline until
lcb_destroy() tears the instance down.  In memory-constrained
containers (Kubernetes pods, tight cgroup limits) this plateau can
be mistaken for a leak and eventually trigger the OOM killer once
multiple instances are stacked in the same process.

Add lcb_trim_memory(lcb_INSTANCE *), marked @UnCommitted.  Calling
it walks every pipeline's nbmgr and reqpool and frees the backing
buffers of blocks on the avail list.  Active (in-flight) blocks are
not touched, connections are not closed, no network I/O is issued.
Intended use is a periodic call from the application's own idle tick
when RSS approaches a configured limit; on a busy instance the call
is a no-op since freed blocks would be reallocated on the next
burst.

Exposed as a dedicated API rather than an lcb_cntl because lcb_cntl
is reachable through the connection string, which is the wrong
surface for an operational command.  A standalone function also
leaves room to evolve the return type (e.g. bytes released) without
another ABI change.

Implementation in two layers:

- netbuf_shrink(nb_MGR *) in src/netbuf/netbuf.c walks each pool's
  avail list, frees the backing buffer of every block, and either
  frees the block header (standalone) or resets the cache-slot
  header so alloc_new_block sees it as free.  Returns bytes
  released.

- lcb_trim_memory() in src/instance.cc iterates every pipeline owned
  by the instance and calls netbuf_shrink on nbmgr and reqpool.

Regression coverage in tests/basic/t_netbuf.cc:

- shrinkFreesAvailBlocksAfterBurst drives 40 concurrent 20 KB
  reservations, releases them in reverse order to populate avail,
  and asserts that shrink empties the list and returns the expected
  byte count.

- shrinkLeavesActiveBlocksAlone holds a reservation, forces a
  transient large allocation onto avail, calls shrink, and verifies
  the held span's buffer is still readable/writable.

- shrinkOnCleanPoolIsNoop calls shrink on a freshly-initialized
  manager and on a cleanly-drained manager and asserts no state
  damage.

All 96 nonio-tests continue to pass.  End-to-end smoke test against
a live cluster (1000 KV stores with 1.5 KB values): invoking
lcb_trim_memory() followed by malloc_trim(0) releases ~340 kB of RSS
on a small run; the amount scales with the peak burst size.

Change-Id: I617bf964cf1329eaff989009a66a02dd57128fd2
Reviewed-on: https://review.couchbase.org/c/libcouchbase/+/243718
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>

Author: avsej

include/libcouchbase/couchbase.h

@@ -2116,6 +2116,52 @@ LIBCOUCHBASE_API
 void lcb_destroy_async(lcb_INSTANCE *instance, const void *arg);
 /**@} (Group: Destroy) */
 
+/**
+ * @brief Release cached pool memory back to the OS.
+ *
+ * libcouchbase services its hot path from per-pipeline buffer pools. When a
+ * packet is released, its backing block is parked on a "free" list inside the
+ * owning pool so the next reservation can be satisfied without calling
+ * `malloc`. The pools grow to match the peak concurrent working set and do
+ * not shrink: a long-lived @c lcb_INSTANCE that briefly bursts and then
+ * idles will retain that peak footprint until it is destroyed.
+ *
+ * Calling this function walks every pipeline owned by @p instance and frees
+ * the backing buffers of blocks sitting on the free list. Active blocks
+ * (those whose memory is currently handed out to an in-flight operation)
+ * are never touched, and no open server connection is closed. After the
+ * call, the next allocation will go back through `malloc`, so the cost of
+ * shrinking only pays off when the process would otherwise hold excess
+ * memory through a quiet period.
+ *
+ * Intended usage is periodic invocation from an application-level idle
+ * tick (e.g. once a minute, or after a batch completes) when the process
+ * runs close to a memory limit. There is no benefit to calling this on a
+ * busy instance — the freed blocks will be reallocated on the next burst.
+ * The call is synchronous and does not issue network I/O.
+ *
+ * Thread safety: this function must be called from the same thread that
+ * owns the I/O loop for @p instance, or with external synchronization that
+ * excludes all other libcouchbase calls on @p instance. It acquires no
+ * internal locks.
+ *
+ * @param instance the instance whose pools should be trimmed. Must be a
+ *                 valid, initialized instance (i.e. one returned by
+ *                 @ref lcb_create and not yet passed to @ref lcb_destroy).
+ *
+ * @uncommitted This function is provided to address operational scenarios
+ *              observed with long-lived instances in memory-constrained
+ *              containers. Its existence and signature may change in a
+ *              future release as the underlying pool implementation
+ *              evolves. Do not rely on it from code that must remain
+ *              compatible across future libcouchbase versions without
+ *              re-validation.
+ *
+ * @see lcb_destroy
+ */
+LIBCOUCHBASE_API
+void lcb_trim_memory(lcb_INSTANCE *instance);
+
 /** @internal */
 #define LCB_DATATYPE_JSON 0x01
 

src/instance.cc

@@ -859,6 +859,25 @@ static void destroy_cb(void *arg)
     lcb_destroy(instance);
 }
 
+LIBCOUCHBASE_API
+void lcb_trim_memory(lcb_INSTANCE *instance)
+{
+    if (instance == nullptr) {
+        return;
+    }
+    mc_CMDQUEUE *cq = &instance->cmdq;
+    nb_SIZE released = 0;
+    for (unsigned ii = 0; ii < cq->_npipelines_ex; ++ii) {
+        mc_PIPELINE *pl = cq->pipelines[ii];
+        if (pl == nullptr) {
+            continue;
+        }
+        released += netbuf_shrink(&pl->nbmgr);
+        released += netbuf_shrink(&pl->reqpool);
+    }
+    lcb_log(LOGARGS(instance, DEBUG), "lcb_trim_memory released %u bytes of idle pool memory", (unsigned)released);
+}
+
 LIBCOUCHBASE_API
 void lcb_destroy_async(lcb_INSTANCE *instance, const void *arg)
 {

src/netbuf/netbuf.c

@@ -774,6 +774,55 @@ void netbuf_cleanup(nb_MGR *mgr)
     mblock_cleanup(&mgr->datapool);
 }
 
+/**
+ * Release the backing buffers of every block in @p pool->avail.
+ *
+ * For cache-slot headers (those owned by @p pool->cacheblocks) the header
+ * itself is left in place so it can be reused by @c alloc_new_block; only
+ * the @c root buffer is freed and the block is reset to its pre-allocation
+ * state. Standalone blocks are freed whole. In either case the block leaves
+ * the @c avail list.
+ *
+ * @return bytes of backing buffer memory released
+ */
+static nb_SIZE mblock_shrink(nb_MBPOOL *pool)
+{
+    sllist_iterator iter;
+    nb_SIZE released = 0;
+
+    SLLIST_ITERFOR(&pool->avail, &iter)
+    {
+        nb_MBLOCK *block = SLLIST_ITEM(iter.cur, nb_MBLOCK, slnode);
+        int is_standalone = mblock_is_standalone(block);
+        nb_SIZE block_size = block->nalloc;
+
+        sllist_iter_remove(&pool->avail, &iter);
+        pool->curblocks--;
+
+        mblock_wipe_block(block);
+
+        if (!is_standalone) {
+            /* Reset the cache-slot so alloc_new_block sees it as free. */
+            block->nalloc = 0;
+            block->start = 0;
+            block->cursor = 0;
+            block->wrap = 0;
+        }
+
+        released += block_size;
+    }
+
+    return released;
+}
+
+nb_SIZE netbuf_shrink(nb_MGR *mgr)
+{
+    nb_SIZE released = 0;
+    released += mblock_shrink(&mgr->datapool);
+    released += mblock_shrink(&mgr->sendq.elempool);
+    return released;
+}
+
 /******************************************************************************
  ******************************************************************************
  ** Block Dumping                                                            **

src/netbuf/netbuf.h

@@ -343,6 +343,22 @@ void netbuf_init(nb_MGR *mgr, const nb_SETTINGS *settings);
 void netbuf_cleanup(nb_MGR *mgr);
 void netbuf_cleanup_packet(nb_MGR *mgr, const void *packet);
 
+/**
+ * @brief Release idle memory back to the system.
+ *
+ * Walks every pool owned by @p mgr and frees the backing buffers of blocks
+ * sitting in the available list (blocks not currently servicing reservations).
+ * The active list is never touched. Cache-slot headers are kept but marked
+ * free so they can be re-used by a subsequent @c netbuf_mblock_reserve.
+ *
+ * Intended for long-lived instances that occasionally burst and then idle.
+ * See CCBC-1685.
+ *
+ * @param mgr the manager to shrink
+ * @return number of bytes released
+ */
+nb_SIZE netbuf_shrink(nb_MGR *mgr);
+
 /**
  * Populates the settings structure with the default settings. This structure
  * may then be modified or tuned and passed to netbuf_init()

tests/basic/t_netbuf.cc

@@ -592,3 +592,100 @@ TEST_F(NetbufTest, testPacketCleanup)
 
     clean_check(&mgr);
 }
+
+/*
+ * Regression tests for CCBC-1685.
+ *
+ * Long-lived lcb instances accumulate netbuf blocks on the avail list after
+ * bursty workloads. Without a way to release them, RSS plateaus at the peak
+ * working set until the instance is destroyed. netbuf_shrink() walks every
+ * pool's avail list and frees the backing buffers, leaving active blocks
+ * untouched.
+ */
+TEST_F(NetbufTest, shrinkFreesAvailBlocksAfterBurst)
+{
+    nb_MGR mgr;
+    netbuf_init(&mgr, nullptr);
+
+    /* Reserve enough independent blocks that multiple go on avail when
+     * released. With the default basealloc of 32 KB, 40 spans of 20 KB each
+     * force a new block every 2 spans. */
+    constexpr int n_spans = 40;
+    constexpr unsigned span_size = 20 * 1024;
+    std::array<nb_SPAN, n_spans> spans{};
+    for (int ii = 0; ii < n_spans; ++ii) {
+        spans[ii].size = span_size;
+        ASSERT_EQ(0, netbuf_mblock_reserve(&mgr, &spans[ii]));
+    }
+
+    /* Release in reverse order so trailing spans flip blocks to empty
+     * quickly and populate the avail list. */
+    for (int ii = n_spans - 1; ii >= 0; --ii) {
+        netbuf_mblock_release(&mgr, &spans[ii]);
+    }
+
+    ASSERT_NE(0, netbuf_is_clean(&mgr));
+    ASSERT_GT(mgr.datapool.curblocks, 0u) << "burst should have populated avail";
+
+    nb_SIZE before_curblocks = mgr.datapool.curblocks;
+    nb_SIZE released = netbuf_shrink(&mgr);
+
+    EXPECT_GE(released, before_curblocks * span_size)
+        << "shrink must return at least the total nalloc of the blocks it frees";
+    EXPECT_EQ(0u, mgr.datapool.curblocks) << "avail list must be empty after shrink";
+
+    /* After shrink, the pool must still be usable. */
+    nb_SPAN span;
+    span.size = 32;
+    ASSERT_EQ(0, netbuf_mblock_reserve(&mgr, &span));
+    netbuf_mblock_release(&mgr, &span);
+
+    clean_check(&mgr);
+}
+
+TEST_F(NetbufTest, shrinkLeavesActiveBlocksAlone)
+{
+    nb_MGR mgr;
+    netbuf_init(&mgr, nullptr);
+
+    /* Reserve a span and hold onto it — active list, not avail. */
+    nb_SPAN active;
+    active.size = 1024;
+    ASSERT_EQ(0, netbuf_mblock_reserve(&mgr, &active));
+
+    /* Cause a separate block to land on avail by allocating and releasing
+     * a larger span. */
+    nb_SPAN transient;
+    transient.size = 64 * 1024;
+    ASSERT_EQ(0, netbuf_mblock_reserve(&mgr, &transient));
+    netbuf_mblock_release(&mgr, &transient);
+
+    nb_SIZE released = netbuf_shrink(&mgr);
+    EXPECT_GT(released, 0u) << "the transient block should be reclaimed";
+
+    /* The active span must still be readable after shrink. */
+    memset(SPAN_BUFFER(&active), 0xAB, active.size);
+    char expected[1024];
+    memset(expected, 0xAB, sizeof(expected));
+    EXPECT_EQ(0, memcmp(SPAN_BUFFER(&active), expected, sizeof(expected)));
+
+    netbuf_mblock_release(&mgr, &active);
+    clean_check(&mgr);
+}
+
+TEST_F(NetbufTest, shrinkOnCleanPoolIsNoop)
+{
+    nb_MGR mgr;
+    netbuf_init(&mgr, nullptr);
+
+    EXPECT_EQ(0u, netbuf_shrink(&mgr));
+
+    /* Cycle a span to confirm the manager is still healthy. */
+    nb_SPAN span;
+    span.size = 500;
+    ASSERT_EQ(0, netbuf_mblock_reserve(&mgr, &span));
+    netbuf_mblock_release(&mgr, &span);
+    EXPECT_NE(0, netbuf_is_clean(&mgr));
+
+    netbuf_cleanup(&mgr);
+}