diff --git a/docs/06-distribution-and-clustering.md b/docs/06-distribution-and-clustering.md
index faf6f4cb..8d08d687 100644
--- a/docs/06-distribution-and-clustering.md
+++ b/docs/06-distribution-and-clustering.md
@@ -166,7 +166,7 @@ High-frequency cc traffic rides `CcMessage` (proto: `proto/cc_request.proto:1407
 
 - **`CcStreamSender`** (`remote/cc_stream_sender.h`): per remote node keeps a *regular* stream and a *long-message* stream (for slow-to-deserialize payloads like `ScanSliceResponse`, so they don't head-of-line-block small messages). A dedicated `connect_thd_` (re)establishes streams (`Connect` RPC of `CcStreamService` then brpc stream accept); each stream carries a version number to serialize concurrent reconnects, plus the peer IP so an IP change forces reconnect (`UpdateStreamIP`). `SendMessageToNg` resolves the NG leader via `ng_leader_cache_` first.
 - **Resend logic**: `SendMessageResult{sent, queued_for_retry, need_reconnect}`. Failed writes are queued per node (`resend_message_list_`, EAGAIN cases in `eagain_resend_*`) and re-driven by `resend_thd_`; a hard failure flags the stream for reconnect. Exception: `SendStandbyMessageToNode` is deliberately best-effort (no retry queue) because standby replication has its own sequence-based resend (see standby doc §4.4).
-- **`CcStreamReceiver`** (`remote/cc_stream_receiver.h`): a `brpc::StreamInputHandler` that parses inbound messages from a pooled `msg_pool_` (zero-alloc steady state) and dispatches in `OnReceiveCcMsg`:
+- **`CcStreamReceiver`** (`remote/cc_stream_receiver.h`): a `brpc::StreamInputHandler` that acquires reusable protobuf shells from `CcMessagePool`, parses inbound messages, and dispatches them in `OnReceiveCcMsg`. Every recycle clears the protobuf before publishing it back to the concurrent pool, so idle messages do not retain payload allocations that the next `ParseFrom` would discard anyway:
   - **Requests** (Acquire/Read/Apply/ScanNext/PostWriteAll/...) are wrapped in pooled `Remote*` cc-request objects (`remote/remote_cc_request.h`, e.g. `RemoteAcquire : AcquireCc`) and enqueued to the proper CcShard — from a shard's point of view a remote request is indistinguishable from a local one (see 03-concurrency-control.md).
   - **Responses** must find their transaction: the message carries `tx_number`, `txm_addr`, `handler_addr`, `command_id`, `tx_term`. The receiver first fences with `CheckLeaderTerm(tx_node_id, tx_term)` (tx node id is encoded in the tx number: `(tx_number >> 32) >> 10`), then dereferences `txm_addr`, takes the txm's **shared forward latch** (`AcquireSharedForwardLatch`), and re-validates `txm->TxNumber() == msg.tx_number && txm->CommandId() == msg.command_id` before touching the `CcHandlerResult` at `handler_addr` (`cc_stream_receiver.cpp:332-400`). The tx_number+command_id check makes stale/duplicate responses for a recycled txm harmless; the latch keeps the txm from being reused mid-update.
   - Locking-protocol requests that block remotely send an **acknowledgement** (`is_ack=true` responses carry the remote cce lock address) so the sender can later probe blocked requests (`BlockedCcReqCheckRequest`).
diff --git a/tx_service/include/remote/cc_message_pool.h b/tx_service/include/remote/cc_message_pool.h
new file mode 100644
index 00000000..654e7ed2
--- /dev/null
+++ b/tx_service/include/remote/cc_message_pool.h
@@ -0,0 +1,77 @@
+/**
+ *    Copyright (C) 2025 EloqData Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under either of the following two licenses:
+ *    1. GNU Affero General Public License, version 3, as published by the Free
+ *    Software Foundation.
+ *    2. GNU General Public License as published by the Free Software
+ *    Foundation; version 2 of the License.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License or GNU General Public License for more
+ *    details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    and GNU General Public License V2 along with this program.  If not, see
+ *    <http://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include <bthread/moodycamelqueue.h>
+
+#include <memory>
+#include <utility>
+
+#include "proto/cc_request.pb.h"
+
+namespace txservice::remote
+{
+/**
+ * @brief Thread-safe reuse pool for CC protobuf messages.
+ *
+ * Recycle clears a message before making it available to another thread. This
+ * keeps the idle pool from retaining payload allocations that ParseFrom would
+ * discard before the next use anyway.
+ */
+class CcMessagePool
+{
+public:
+    CcMessagePool() = default;
+    CcMessagePool(const CcMessagePool &) = delete;
+    CcMessagePool &operator=(const CcMessagePool &) = delete;
+    CcMessagePool(CcMessagePool &&) = delete;
+    CcMessagePool &operator=(CcMessagePool &&) = delete;
+
+    /**
+     * @brief Returns an empty pooled message or allocates a new one.
+     *
+     * The caller owns the returned message until passing it to Recycle().
+     */
+    std::unique_ptr<CcMessage> Acquire()
+    {
+        std::unique_ptr<CcMessage> msg;
+        if (!pool_.try_dequeue(msg))
+        {
+            msg = std::make_unique<CcMessage>();
+        }
+        return msg;
+    }
+
+    /**
+     * @brief Clears and transfers a message back to the shared pool.
+     *
+     * @param msg A non-null message exclusively owned by the caller.
+     */
+    void Recycle(std::unique_ptr<CcMessage> msg)
+    {
+        msg->Clear();
+        pool_.enqueue(std::move(msg));
+    }
+
+private:
+    moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> pool_;
+};
+}  // namespace txservice::remote
diff --git a/tx_service/include/remote/cc_stream_receiver.h b/tx_service/include/remote/cc_stream_receiver.h
index 1b660bdf..20ed7cba 100644
--- a/tx_service/include/remote/cc_stream_receiver.h
+++ b/tx_service/include/remote/cc_stream_receiver.h
@@ -31,6 +31,7 @@
 
 #include "cc_req_pool.h"
 #include "proto/cc_request.pb.h"
+#include "remote/cc_message_pool.h"
 #include "tx_record.h"
 #include "type.h"
 
@@ -43,9 +44,7 @@ namespace remote
 class CcStreamReceiver : public brpc::StreamInputHandler, public CcStreamService
 {
 public:
-    explicit CcStreamReceiver(
-        LocalCcShards &local_shards,
-        moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool);
+    CcStreamReceiver(LocalCcShards &local_shards, CcMessagePool &msg_pool);
     ~CcStreamReceiver() = default;
 
     void Shutdown();
@@ -82,7 +81,7 @@ class CcStreamReceiver : public brpc::StreamInputHandler, public CcStreamService
     // receives a message, de-serializes it and dispatches it to local shards
     // for processing. The message is put back into the pool after the cc
     // request is processed.
-    moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool_;
+    CcMessagePool &msg_pool_;
     moodycamel::ConcurrentQueue<std::unique_ptr<ScanSliceResponse>>
         scan_resp_pool_;
 };
diff --git a/tx_service/include/remote/cc_stream_sender.h b/tx_service/include/remote/cc_stream_sender.h
index 5bb2f1d3..41a82a18 100644
--- a/tx_service/include/remote/cc_stream_sender.h
+++ b/tx_service/include/remote/cc_stream_sender.h
@@ -34,6 +34,7 @@
 
 #include "cc/cc_handler_result.h"
 #include "proto/cc_request.pb.h"
+#include "remote/cc_message_pool.h"
 #include "sharder.h"
 
 namespace txservice
@@ -96,8 +97,7 @@ struct SendMessageResult
 class CcStreamSender
 {
 public:
-    CcStreamSender(
-        moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool);
+    explicit CcStreamSender(CcMessagePool &msg_pool);
     ~CcStreamSender();
 
     void RecycleCcMsg(std::unique_ptr<CcMessage> msg);
@@ -169,7 +169,7 @@ class CcStreamSender
     int ConnectStream(uint32_t node_id, int64_t version);
     int ConnectLongMsgStream(uint32_t node_id, int64_t version);
 
-    moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool_;
+    CcMessagePool &msg_pool_;
 
     brpc::StreamWriteOptions stream_write_options_;
 
diff --git a/tx_service/include/sharder.h b/tx_service/include/sharder.h
index 77f60fda..52ae9d2e 100644
--- a/tx_service/include/sharder.h
+++ b/tx_service/include/sharder.h
@@ -37,6 +37,7 @@
 #include "brpc/server.h"
 #include "butil/third_party/murmurhash3/murmurhash3.h"
 #include "proto/cc_request.pb.h"
+#include "remote/cc_message_pool.h"
 #include "tx_serialize.h"
 #include "txlog.h"
 #include "type.h"
@@ -839,7 +840,7 @@ class Sharder
      */
     std::atomic<bool> cc_nodes_init_{false};
 
-    moodycamel::ConcurrentQueue<std::unique_ptr<remote::CcMessage>> msg_pool_;
+    remote::CcMessagePool msg_pool_;
 
     // The cc stream sender establishes connections to remote nodes and
     // sends cc requests and responses to remote nodes via streams. It is
diff --git a/tx_service/src/remote/cc_stream_receiver.cpp b/tx_service/src/remote/cc_stream_receiver.cpp
index 3858074a..d7aa9782 100644
--- a/tx_service/src/remote/cc_stream_receiver.cpp
+++ b/tx_service/src/remote/cc_stream_receiver.cpp
@@ -110,9 +110,8 @@ thread_local CcRequestPool<KeyObjectStandbyForwardCc>
     key_obj_standby_forward_pool_;
 thread_local CcRequestPool<ParseCcMsgCc> parse_standby_forward_pool_;
 
-CcStreamReceiver::CcStreamReceiver(
-    LocalCcShards &local_shards,
-    moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool)
+CcStreamReceiver::CcStreamReceiver(LocalCcShards &local_shards,
+                                   CcMessagePool &msg_pool)
     : local_shards_(local_shards), msg_pool_(msg_pool)
 {
 }
@@ -278,7 +277,7 @@ int CcStreamReceiver::on_received_messages(brpc::StreamId stream_id,
                     remote::CcMessage::MessageType::
                         CcMessage_MessageType_KeyObjectStandbyForwardRequest)
                 {
-                    msg_pool_.enqueue(std::move(cc_msg));
+                    msg_pool_.Recycle(std::move(cc_msg));
                     continue;
                 }
                 OnReceiveCcMsg(std::move(cc_msg));
@@ -326,15 +325,7 @@ void CcStreamReceiver::on_closed(brpc::StreamId stream)
 
 std::unique_ptr<CcMessage> CcStreamReceiver::GetCcMsg()
 {
-    std::unique_ptr<CcMessage> msg;
-    if (msg_pool_.try_dequeue(msg))
-    {
-        return msg;
-    }
-    else
-    {
-        return std::make_unique<CcMessage>();
-    }
+    return msg_pool_.Acquire();
 }
 
 std::unique_ptr<ScanSliceResponse> CcStreamReceiver::GetScanSliceResp()
@@ -493,7 +484,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -512,7 +503,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -525,7 +516,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         if (acq_res.IsRemoteHdResultSet(std::memory_order_acquire))
         {
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             txm->ReleaseSharedForwardLatch();
             break;
         }
@@ -583,7 +574,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             txm->ReleaseSharedForwardLatch();
         }
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_AcquireAllRequest:
@@ -606,7 +597,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -623,7 +614,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -685,7 +676,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             txm->ReleaseSharedForwardLatch();
         }
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ValidateRequest:
@@ -725,7 +716,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 Sharder::Instance().GetCcStreamSender();
 
             cc_stream_sender->SendMessageToNode(req.src_node_id(), return_msg);
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
         }
         else
         {
@@ -749,7 +740,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -767,7 +758,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -791,7 +782,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         txm->ReleaseSharedForwardLatch();
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_PostprocessResponse:
@@ -806,7 +797,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -824,7 +815,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -845,7 +836,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         txm->ReleaseSharedForwardLatch();
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ReadRequest:
@@ -880,7 +871,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -900,7 +891,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -950,7 +941,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
             txm->ReleaseSharedForwardLatch();
         }
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_PostCommitRequest:
@@ -990,7 +981,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 Sharder::Instance().GetCcStreamSender();
             cc_stream_sender->SendMessageToNode(post_commit.src_node_id(),
                                                 return_msg);
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
         }
         else
         {
@@ -1026,7 +1017,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 Sharder::Instance().GetCcStreamSender();
             cc_stream_sender->SendMessageToNode(post_commit.src_node_id(),
                                                 return_msg);
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
         }
         else
         {
@@ -1050,7 +1041,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
     {
         assert(false && "Unimplemented");
         LOG(ERROR) << "ScanOpenRequest is unsupported; replying error.";
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ScanOpenResponse:
@@ -1064,7 +1055,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         assert(false && "Unimplemented");
         LOG(ERROR) << "ScanOpenResponse is unsupported; replying error.";
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ScanNextRequest:
@@ -1098,7 +1089,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -1116,7 +1107,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1137,7 +1128,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             // The handler result has been reset while this response is in
             // transit. Drops this stale response to avoid dereferencing a
             // recycled scanner pointer.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             txm->ReleaseSharedForwardLatch();
             break;
         }
@@ -1278,7 +1269,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         txm->ReleaseSharedForwardLatch();
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ScanSliceRequest:
@@ -1310,7 +1301,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         int64_t tx_term = msg->tx_term();
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -1328,7 +1319,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1347,7 +1338,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         }
 
         txm->ReleaseSharedForwardLatch();
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_FaultInjectRequest:
@@ -1370,7 +1361,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         CcHandlerResult<bool> *hd_res =
@@ -1387,7 +1378,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             // The original tx has terminated and the tx machine has been
             // recycled. The response message is directed to an obsolete tx.
             // Skips setting the cc handler result.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             txm->ReleaseSharedForwardLatch();
             break;
         }
@@ -1406,7 +1397,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         }
         txm->ReleaseSharedForwardLatch();
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_AnalyzeTableAllRequest:
@@ -1429,7 +1420,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         int64_t tx_term = msg->tx_term();
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -1447,7 +1438,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1467,7 +1458,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         }
 
         txm->ReleaseSharedForwardLatch();
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::
@@ -1492,7 +1483,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         CcHandlerResult<bool> *hd_res =
@@ -1511,7 +1502,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             hd_res->SetFinished();
         }
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_RecoverStateCheckRequest:
@@ -1545,7 +1536,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
 
         Sharder::Instance().GetCcStreamSender()->SendMessageToNode(
             req.src_node_id(), send_msg);
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::
@@ -1565,7 +1556,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                        << ", error_code:" << resp.error_code();
             Sharder::Instance().UpdateLeader(resp.node_group_id());
         }
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_DeadLockRequest:
@@ -1584,7 +1575,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
     {
         const DeadLockResponse &rsp = msg->dead_lock_response();
         DeadLockCheck::MergeRemoteWaitingLockInfo(&rsp);
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::
@@ -1612,7 +1603,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
     }
     case CcMessage::MessageType::CcMessage_MessageType_AbortTransactionResponse:
     {
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_BlockedCcReqCheckRequest:
@@ -1640,7 +1631,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
 
@@ -1664,7 +1655,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1674,7 +1665,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             if (acq_key_result_vec.at(resp.acq_key_result_vec_idx())
                     .IsRemoteHdResultSet(std::memory_order_acquire))
             {
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1716,7 +1707,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1745,7 +1736,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             txm->ReleaseSharedForwardLatch();
         }
 
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_KickoutDataRequest:
@@ -1787,7 +1778,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         {
             // The tx coordinator node has failed. Pointer stability
             // does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             LOG(ERROR) << "Receive remote kickoutccentry response, but tx"
                           " coordinator has failed.";
             break;
@@ -1807,7 +1798,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx state machine has
                 // been recycled. The response message is directed to an
                 // obsolete tx.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1830,7 +1821,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         txm->ReleaseSharedForwardLatch();
 
         // Recycle the cc message
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_ApplyRequest:
@@ -1853,7 +1844,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
             // The tx node has failed. Pointer stability does not hold anymore.
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -1871,7 +1862,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
                 // The original tx has terminated and the tx machine has been
                 // recycled. The response message is directed to an obsolete tx.
                 // Skips setting the cc handler result.
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -1905,7 +1896,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             }
         }
         txm->ReleaseSharedForwardLatch();
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::CcMessage_MessageType_PublishRequest:
@@ -1953,7 +1944,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         }
 
         dbcc->AddRemoteObjSize(resp.dbsize_term(), total_obj_sizes);
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     case CcMessage::MessageType::
@@ -1992,7 +1983,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         int64_t tx_term = msg->tx_term();
         if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term))
         {
-            msg_pool_.enqueue(std::move(msg));
+            msg_pool_.Recycle(std::move(msg));
             break;
         }
         else
@@ -2007,7 +1998,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
             if (txm->TxNumber() != msg->tx_number() ||
                 txm->CommandId() != msg->command_id())
             {
-                msg_pool_.enqueue(std::move(msg));
+                msg_pool_.Recycle(std::move(msg));
                 txm->ReleaseSharedForwardLatch();
                 break;
             }
@@ -2027,7 +2018,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr<CcMessage> msg)
         }
 
         txm->ReleaseSharedForwardLatch();
-        msg_pool_.enqueue(std::move(msg));
+        msg_pool_.Recycle(std::move(msg));
         break;
     }
     default:
diff --git a/tx_service/src/remote/cc_stream_sender.cpp b/tx_service/src/remote/cc_stream_sender.cpp
index c28237c9..b52dc4ec 100644
--- a/tx_service/src/remote/cc_stream_sender.cpp
+++ b/tx_service/src/remote/cc_stream_sender.cpp
@@ -52,8 +52,7 @@ CcStreamSender::~CcStreamSender()
     resend_thd_.join();
 }
 
-CcStreamSender::CcStreamSender(
-    moodycamel::ConcurrentQueue<std::unique_ptr<CcMessage>> &msg_pool)
+CcStreamSender::CcStreamSender(CcMessagePool &msg_pool)
     : msg_pool_(msg_pool), terminate_(false), to_connect_flag_(false)
 {
     stream_write_options_.write_in_background = true;
@@ -65,7 +64,7 @@ CcStreamSender::CcStreamSender(
 
 void CcStreamSender::RecycleCcMsg(std::unique_ptr<CcMessage> msg)
 {
-    msg_pool_.enqueue(std::move(msg));
+    msg_pool_.Recycle(std::move(msg));
 }
 
 void CcStreamSender::ReConnectStream(uint32_t dest_node_id)
diff --git a/tx_service/tests/CMakeLists.txt b/tx_service/tests/CMakeLists.txt
index cec3ae45..66898a88 100644
--- a/tx_service/tests/CMakeLists.txt
+++ b/tx_service/tests/CMakeLists.txt
@@ -59,6 +59,7 @@ set_property(TARGET txnode PROPERTY CXX_STANDARD 20)
 # files). They link Catch2::Catch2WithMain.
 set(CATCH_MAIN_TESTS
     CcEntry-Test
+    CcMessagePool-Test
     CcPage-Test
     LargeObjLRU-Test
     CcRequestWait-Test
diff --git a/tx_service/tests/CcMessagePool-Test.cpp b/tx_service/tests/CcMessagePool-Test.cpp
new file mode 100644
index 00000000..74cd8b77
--- /dev/null
+++ b/tx_service/tests/CcMessagePool-Test.cpp
@@ -0,0 +1,57 @@
+/**
+ *    Copyright (C) 2025 EloqData Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under either of the following two licenses:
+ *    1. GNU Affero General Public License, version 3, as published by the Free
+ *    Software Foundation.
+ *    2. GNU General Public License as published by the Free Software
+ *    Foundation; version 2 of the License.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License or GNU General Public License for more
+ *    details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    and GNU General Public License V2 along with this program.  If not, see
+ *    <http://www.gnu.org/licenses/>.
+ */
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <utility>
+
+/** Let Catch provide main(). */
+#include <catch2/catch_all.hpp>
+
+#include "remote/cc_message_pool.h"
+
+namespace txservice::remote
+{
+TEST_CASE("CcMessagePool clears payloads before recycling", "[cc-message-pool]")
+{
+    constexpr size_t payload_size = 64 * 1024;
+
+    CcMessagePool pool;
+    std::unique_ptr<CcMessage> msg = pool.Acquire();
+    CcMessage *original = msg.get();
+    msg->set_type(CcMessage::KeyObjectStandbyForwardRequest);
+    msg->set_tx_number(42);
+    KeyObjectStandbyForwardRequest *request =
+        msg->mutable_key_obj_standby_forward_req();
+    request->set_key(std::string(payload_size, 'k'));
+    request->add_cmd_list(std::string(payload_size, 'c'));
+    const size_t populated_space = msg->SpaceUsedLong();
+
+    pool.Recycle(std::move(msg));
+    std::unique_ptr<CcMessage> recycled = pool.Acquire();
+
+    REQUIRE(recycled.get() == original);
+    REQUIRE(recycled->content_case() == CcMessage::CONTENT_NOT_SET);
+    REQUIRE(recycled->type() == CcMessage::AcquireRequest);
+    REQUIRE(recycled->tx_number() == 0);
+    REQUIRE(recycled->SpaceUsedLong() < populated_space / 2);
+}
+}  // namespace txservice::remote