fix(buffer)!: transfer receive storage with move-only ownership

docs/reference/directory-structure.md

@@ -34,6 +34,8 @@ The MySQL/Redis interfaces have one source each: `src/plugins/mysql/mysql_module
 
 Core has one source copy under `src/core/`. `ark::core` maps its explicit public headers into the selected build tree, preserving `<ark/plugin_manager.hpp>` and nested include paths. Edit source headers rather than generated copies. The `app`/`app_d` executable and manifest format are unchanged.
 
+`<ark/buffer.hpp>` retains its public path with move-only ownership; rebuild affected Core/Net consumers for the source/layout change. See [receive-buffer ownership](network.md#receive-buffer-ownership) for migration, allocation-hook compatibility and focused tests.
+
 ## CMake targets
 
 | Target | Usage |

docs/reference/directory-structure_CN.md

@@ -32,6 +32,8 @@ MySQL、Redis 的唯一公开接口源码分别是 `src/plugins/mysql/mysql_modu
 
 核心头文件只有 `src/core/` 中的一份源码。`ark::core` 将显式公开清单映射到选定构建目录，保留 `<ark/plugin_manager.hpp>` 等包含路径。修改源码头文件，不编辑生成副本。启动程序仍为 `app`／`app_d`，清单格式保持不变。
 
+`<ark/buffer.hpp>` 保留原有公开路径，所有权改为仅支持移动；源码／布局发生变化，需要重建受影响的 Core／Net 消费者。迁移、分配钩子兼容性与定向测试见[收包缓冲的所有权](network_CN.md#收包缓冲的所有权)。
+
 ## CMake 目标
 
 | 目标 | 用途 |

docs/reference/network.md

@@ -95,6 +95,27 @@ TCP, WebSocket and RUDP endpoints use a 4 KiB initial application receive buffer
 
 This is the framing buffer for application messages, configured separately from kernel socket send/receive buffers. It is not a 4 KiB maximum message size. The [local receive-buffer report](../performance/receive-buffers.md) retains the allocation, fragmented-input and large-batch throughput tradeoff. It does not establish network RTT or connection capacity.
 
+## Receive-buffer ownership
+
+**Breaking source/layout change:** `<ark/buffer.hpp>` exposes a move-only `ark::buffer`. Copy construction and copy assignment are deleted to prevent two objects from deleting the same array. Rebuild the Core/Net consumers and their plugins together. The tracked Net sessions already prohibit copying and own their buffer in place; copying a `shared_ptr` session handle does not copy that buffer. This repository audit does not establish whether external consumers relied on the former implicit copy operations.
+
+Move construction and move assignment are `noexcept` and transfer the allocation, unread bytes, capacity and cursor state. Assignment releases the destination's previous allocation; self-move leaves it unchanged. A moved-from buffer has zero capacity, zero unread bytes and a null data pointer; it can be destroyed, assigned or written to again. Existing public method signatures remain available. Standalone construction still defaults to 512 KiB, while Net sessions retain their separately configured 4 KiB default.
+
+```cpp
+ark::buffer pending(4096);
+pending.write("abc", 3);
+ark::buffer processing(std::move(pending)); // transfers ownership
+pending.write("next", 4);                 // allocates fresh storage
+```
+
+Include `<utility>` for `std::move`. `get_data()` returns a borrowed pointer into the current allocation; keep its owner alive and reacquire the pointer after compaction, growth or ownership replacement. This change introduces no clone API. Applications that need independent data can explicitly write the readable bytes into another buffer while the source remains valid.
+
+The buffer keeps `ARK_NEW_ARRAY` / `ARK_DELETE_ARRAY` as its narrow allocation compatibility boundary, now using the same pair for construction, growth and release. This corrects the previous growth path that bypassed custom hooks with raw `new[]` / `delete[]`. Define a compatible, nonthrowing release hook consistently across translation units before including the header. Without overrides, allocation still uses throwing `new[]`: `std::bad_alloc` propagates and failed growth retains the old bytes, capacity and cursors. If a custom allocator returns null, construction retains an empty zero-capacity buffer and `write()` returns `false` without changing the existing state.
+
+A zero-length write succeeds without dereferencing its pointer; a null source with nonzero length returns `false`. Removing more than the unread length is a no-op, including an overflowing count. Unrepresentable growth throws `std::length_error` before allocation. Normal reads, writes, compaction and capacity growth retain their existing behavior.
+
+With `ARK_BUILD_CORE_TESTS=ON`, `check-buffer` runs the lifetime/move, compaction/growth, boundary, deterministic default-allocation failure and separate custom-hook regressions; it is also part of `check-core`. The allocation-failure test fails one small array request instead of requesting excessive memory. `check-network` retains the existing Net framing/fragmented-input coverage. These checks do not establish service or capacity acceptance.
+
 ## Member callbacks and borrowed arguments
 
 Endpoint constructors and client/server service registration bind the supplied

docs/reference/network_CN.md

@@ -80,6 +80,27 @@ TCP、WebSocket 与 RUDP 端点使用 4 KiB 初始应用层收包缓冲，随后
 
 这是应用消息的拆包缓冲，与内核 socket 收发缓冲分别配置，也不是 4 KiB 的最大消息限制。[本地收包缓冲报告](../performance/receive-buffers_CN.md)保留了内存申请、分片输入与大批量吞吐量之间的取舍，不把这些结果视为网络 RTT 或连接容量。
 
+## 收包缓冲的所有权
+
+**源码／布局不兼容变更：** `<ark/buffer.hpp>` 提供的 `ark::buffer` 仅支持移动。复制构造和复制赋值被显式删除，避免两个对象重复释放同一数组。Core／Net 消费者及其插件必须一起重建。仓库中的 Net Session 原本就禁止复制，并在对象内部持有缓冲；复制 `shared_ptr` Session 句柄不会复制该缓冲。这次仓库调用检查不能证明外部消费者从未使用原有隐式复制操作。
+
+移动构造和移动赋值均为 `noexcept`，转移数组、未读数据、容量和游标状态。移动赋值释放目标原有数组；自身移动保持原状态。移出对象的容量和未读长度均为零，数据指针为空，可继续销毁、赋值或写入。原有公开方法签名继续保留。独立构造默认容量仍为 512 KiB，Net Session 继续使用独立配置的 4 KiB 默认容量。
+
+```cpp
+ark::buffer pending(4096);
+pending.write("abc", 3);
+ark::buffer processing(std::move(pending)); // 转移所有权
+pending.write("next", 4);                 // 重新申请存储
+```
+
+使用 `std::move` 时包含 `<utility>`。`get_data()` 返回当前数组中的借用指针；所属对象必须保持有效，在压缩、扩容或替换所有权后重新获取指针。本次不增加 clone API。需要独立数据的应用可在源数据有效期间，将可读字节显式写入另一个缓冲。
+
+缓冲通过 `ARK_NEW_ARRAY`／`ARK_DELETE_ARRAY` 保留最小分配兼容边界，构造、扩容和释放现在使用同一对钩子。这修正了原扩容路径通过裸 `new[]`／`delete[]` 绕过自定义钩子的问题。覆盖钩子时，须在包含头文件前定义匹配且释放时不抛异常的实现，并确保各编译单元定义一致。默认分配仍使用会抛异常的 `new[]`：`std::bad_alloc` 继续向外传播，扩容失败保留原有数据、容量和游标。自定义分配器返回空指针时，构造得到容量为零的空缓冲；`write()` 返回 `false`，不改变已有状态。
+
+零长度写入成功且不解引用传入指针；非零长度的空源指针返回 `false`。移除数量超过未读长度时不改变状态，包括会使加法溢出的数量。容量无法表示时，在分配前抛出 `std::length_error`。正常读写、压缩和容量增长行为保持不变。
+
+开启 `ARK_BUILD_CORE_TESTS=ON` 后，`check-buffer` 运行生命周期／移动、压缩／扩容、边界、默认分配确定性失败及独立自定义钩子回归，并纳入 `check-core`。分配失败测试只使一次小数组申请失败，不申请巨量内存。`check-network` 保留既有 Net 拆包与分片输入覆盖。这些检查不代表服务或容量验收。
+
 ## 成员回调与借用参数
 
 端点构造函数与客户端／服务端服务的注册接口通过 C++20 `std::bind_front`，

src/core/buffer.hpp

@@ -21,66 +21,86 @@
 #pragma once
 
 #include <algorithm>
+#include <cstring>
 #include <limits>
+#include <memory>
 #include <stdexcept>
+#include <utility>
 
 #include "ark/macros.hpp"
 
 namespace ark {
 
 class buffer final
 {
+    struct data_deleter
+    {
+        void operator()(char* value) const noexcept {
+            ARK_DELETE_ARRAY(char, value);
+        }
+    };
+
+    using owned_data = std::unique_ptr<char[], data_deleter>;
+
+    static char* allocate_data(size_t size) {
+        return ARK_NEW_ARRAY(char, size);
+    }
+
 public:
     explicit buffer(size_t buffer_size = 1024 * 512) {
-        data_ = ARK_NEW_ARRAY(char, buffer_size);
+        data_.reset(allocate_data(buffer_size));
         if (data_ != nullptr)
         {
             data_size_ = buffer_size;
         }
     }
 
-    ~buffer() {
-        ARK_DELETE_ARRAY(char, data_);
+    ~buffer() = default;
+
+    buffer(const buffer&) = delete;
+    buffer& operator=(const buffer&) = delete;
+
+    buffer(buffer&& other) noexcept
+        : data_(std::move(other.data_))
+        , data_size_(std::exchange(other.data_size_, 0))
+        , write_pos_(std::exchange(other.write_pos_, 0))
+        , read_pos_(std::exchange(other.read_pos_, 0)) {
+    }
+
+    buffer& operator=(buffer&& other) noexcept {
+        if (this != &other)
+        {
+            data_ = std::move(other.data_);
+            data_size_ = std::exchange(other.data_size_, 0);
+            write_pos_ = std::exchange(other.write_pos_, 0);
+            read_pos_ = std::exchange(other.read_pos_, 0);
+        }
+        return *this;
     }
 
     size_t get_capacity() const noexcept {
         return data_size_;
     }
 
     bool write(const char* data, size_t len) {
-        bool write_ret = true;
-
-        if (get_write_valid_count() >= len)
+        if (len == 0)
         {
-            memcpy(get_write_ptr(), data, len);
-            add_write_pos(len);
+            return true;
         }
-        else
+        if (data == nullptr)
         {
-            size_t left_len = data_size_ - get_length();
-
-            if (left_len >= len)
-            {
-                adjust_to_head();
-                write(data, len);
-            }
-            else
+            return false;
+        }
+        if (get_write_valid_count() < len) [[unlikely]]
+        {
+            if (!prepare_write(len))
             {
-                size_t need_len = len - left_len;
-
-                if (need_len > 0)
-                {
-                    grow(need_len);
-                    write(data, len);
-                }
-                else
-                {
-                    write_ret = false;
-                }
+                return false;
             }
         }
-
-        return write_ret;
+        std::memcpy(get_write_ptr(), data, len);
+        add_write_pos(len);
+        return true;
     }
 
     size_t get_length() {
@@ -90,7 +110,7 @@ class buffer final
     char* get_data() {
         if (read_pos_ < data_size_)
         {
-            return data_ + read_pos_;
+            return data_.get() + read_pos_;
         }
         else
         {
@@ -99,15 +119,23 @@ class buffer final
     }
 
     void remove_data(size_t value) {
-        size_t temp = read_pos_ + value;
-
-        if (temp <= data_size_)
+        if (value <= get_length())
         {
-            read_pos_ = temp;
+            read_pos_ += value;
         }
     }
 
 private:
+    bool prepare_write(size_t len) {
+        const size_t left_len = data_size_ - get_length();
+        if (left_len >= len)
+        {
+            adjust_to_head();
+            return true;
+        }
+        return grow(len - left_len);
+    }
+
     void adjust_to_head() {
         size_t len = 0;
 
@@ -120,7 +148,7 @@ class buffer final
 
         if (len > 0)
         {
-            memmove(data_, data_ + read_pos_, len);
+            std::memmove(data_.get(), data_.get() + read_pos_, len);
         }
 
         read_pos_ = 0;
@@ -156,34 +184,40 @@ class buffer final
     char* get_write_ptr() {
         if (write_pos_ < data_size_)
         {
-            return data_ + write_pos_;
+            return data_.get() + write_pos_;
         }
         else
         {
             return nullptr;
         }
     }
 
-    void grow(size_t len) {
+    bool grow(size_t len) {
         if (len > std::numeric_limits<size_t>::max() - data_size_)
         {
             throw std::length_error("Receive buffer size exceeds the addressable range");
         }
         const size_t required = data_size_ + len;
         const size_t doubled = data_size_ <= std::numeric_limits<size_t>::max() / 2 ? data_size_ * 2 : required;
         const size_t n = std::max(required, doubled);
-        char* d = new char[n];
-
+        owned_data candidate(allocate_data(n));
+        if (!candidate)
+        {
+            return false;
+        }
         const size_t unread = get_length();
-        memcpy(d, data_ + read_pos_, unread);
+        if (unread > 0)
+        {
+            std::memcpy(candidate.get(), data_.get() + read_pos_, unread);
+        }
+        data_ = std::move(candidate);
         read_pos_ = 0;
         write_pos_ = unread;
         data_size_ = n;
-        delete[] data_;
-        data_ = d;
+        return true;
     }
 
-    char* data_{nullptr};
+    owned_data data_;
     size_t data_size_{0};
 
     size_t write_pos_{0};

tests/app/business_runtime.cmake

@@ -44,11 +44,18 @@ foreach(mode drain drain_deadline drain_max_deadline pre_shut_false shut_false s
     endif()
     file(STRINGS "${directory}/business-runtime.trace" events)
     set(previous -1)
-    set(expected_events start-1 root-rejected)
+    if(mode STREQUAL "timeout_native")
+        set(expected_events native-pending root-rejected owner-stop native-exit)
+        if("start-1" IN_LIST events OR "complete-begin" IN_LIST events)
+            message(FATAL_ERROR "${mode}: native-readiness case must not wait for managed tasks: ${events}")
+        endif()
+    else()
+        set(expected_events start-1 root-rejected)
+    endif()
     if(mode MATCHES "^update_throw")
         list(APPEND expected_events update-throw update-resumed)
     endif()
-    if(NOT mode STREQUAL "timeout_stalled")
+    if(NOT mode STREQUAL "timeout_stalled" AND NOT mode STREQUAL "timeout_native")
         list(APPEND expected_events reply-1 received-1 start-2 reply-2 received-2 complete-begin callback-stop-pending complete-end owner-stop native-exit)
     endif()
     if(mode MATCHES "^drain" OR mode MATCHES "^(pre_shut_false|shut_false|shut_throw)$")

tests/app/business_runtime_plugin.cpp

@@ -27,6 +27,8 @@ void trace(const std::string& event) {
 
 // A controlled protocol completion, not a real database client or public plugin.
 // No result is released until the real launcher enters managed-task drain.
+// The native-readiness deadline case has no task dependency: scheduling the
+// coroutine chain must not consume its separate 50 ms shutdown deadline.
 class probe_module final : public ark::module
 {
 public:
@@ -54,6 +56,11 @@ class probe_module final : public ark::module
             native_exited_.store(true, std::memory_order_release);
         });
         auto& tasks = manager_.create_task_group(*this, 2);
+        if (mode_ == "timeout_native")
+        {
+            trace("native-pending");
+            return true;
+        }
         return tasks.start(sequence(this), [this](std::exception_ptr error) {
             if (error)
             {
@@ -101,7 +108,7 @@ class probe_module final : public ark::module
         }
         const auto delayed =
             draining && std::chrono::steady_clock::now() - drain_started_ < std::chrono::milliseconds(200);
-        if (completion_returned_ && !owner_stopped_)
+        if ((completion_returned_ || (draining && mode_ == "timeout_native")) && !owner_stopped_)
         {
             owner_stopped_ = true;
             trace("owner-stop");
@@ -110,19 +117,19 @@ class probe_module final : public ark::module
         {
             native_release_.store(true, std::memory_order_release);
         }
-        if (pending_)
+        if ((pending_ || mode_ == "timeout_native") && !stop_requested_ && !draining)
         {
-            const auto state = manager_.get_shutdown_state();
-            if (state == ark::plugin_manager::shutdown_state::running && !stop_requested_)
-            {
-                stop_requested_ = true;
-                trace("request-stop");
+            stop_requested_ = true;
+            trace("request-stop");
 #if !defined(_WIN32)
-                std::raise(SIGTERM);
+            std::raise(SIGTERM);
 #endif
-            }
-            else if (state == ark::plugin_manager::shutdown_state::draining && mode_ != "timeout_stalled" &&
-                     (mode_ != "timeout_task" || !delayed))
+        }
+        if (pending_)
+        {
+            const auto state = manager_.get_shutdown_state();
+            if (state == ark::plugin_manager::shutdown_state::draining && mode_ != "timeout_stalled" &&
+                (mode_ != "timeout_task" || !delayed))
             {
                 trace("reply-" + std::to_string(started_));
                 auto complete = std::move(pending_);

tests/core/CMakeLists.txt

@@ -32,6 +32,21 @@ add_custom_target(check-logging
     USES_TERMINAL)
 add_dependencies(check-core check-logging)
 
+# Separate binaries keep custom allocation hooks and new[] replacement isolated.
+add_executable(ark_buffer_tests main.cpp buffers.cpp)
+set_property(TARGET ark_buffer_tests PROPERTY LINK_LIBRARIES ark::core)
+add_executable(ark_buffer_failure_tests main.cpp buffer_failure.cpp)
+set_property(TARGET ark_buffer_failure_tests PROPERTY LINK_LIBRARIES ark::core)
+add_executable(ark_buffer_hook_tests main.cpp buffer_hooks.cpp)
+set_property(TARGET ark_buffer_hook_tests PROPERTY LINK_LIBRARIES ark::core)
+add_custom_target(check-buffer
+    COMMAND $<TARGET_FILE:ark_buffer_tests>
+    COMMAND $<TARGET_FILE:ark_buffer_failure_tests>
+    COMMAND $<TARGET_FILE:ark_buffer_hook_tests>
+    DEPENDS ark_buffer_tests ark_buffer_failure_tests ark_buffer_hook_tests
+    USES_TERMINAL VERBATIM)
+add_dependencies(check-core check-buffer)
+
 find_package(Threads REQUIRED)
 add_library(ark_task_header_checks OBJECT headers/task.cpp)
 # This SDK consumer uses the public core target without aggregate plugin APIs.