fix(fec): release incomplete groups after staging exceptions

docs/features/rudp/fec.md

@@ -70,6 +70,7 @@ client.set_fec_options(settings);
 | Burst removes more than 2 shards | ARQ is still required |
 | Effective parity | Byte budgets and minimum group size may reduce it below configuration |
 | Group retention | Complete groups release immediately; incomplete groups expire |
+| Shard staging allocation failure | `fec::transport::input` releases the affected incomplete recovery group and propagates the exception; application delivery callback exceptions preserve accepted state |
 | Retention full | Skip extra FEC buffering; valid systematic data still reaches ARQ |
 | Duplicate filtering | Fixed 4096-group window |
 | Coding scratch space | Bounded by shard count and MTU, separate from retention budget |

docs/features/rudp/fec_CN.md

@@ -69,6 +69,7 @@ client.set_fec_options(settings);
 | 突发丢失超过 2 个分片 | 仍需 ARQ |
 | 实际校验数 | 字节预算与最小分组大小可能使其低于配置值 |
 | 组保留期限 | 完整组立即释放，未完成组到期释放 |
+| 分片暂存分配失败 | `fec::transport::input` 释放受影响的未完成恢复组并继续抛出异常；应用交付回调抛异常时保留已接收状态 |
 | 缓存不足 | 跳过额外 FEC 保留；合法原始数据继续进入 ARQ，可靠交付仍生效 |
 | 重复过滤 | 固定 4096 组窗口 |
 | 编解码临时空间 | 受分片数与 MTU 限制，在保留预算之外另计 |

include/arknet/fec/transport.hpp

@@ -7,6 +7,7 @@
 #include <cmath>
 #include <cstddef>
 #include <cstdint>
+#include <exception>
 #include <functional>
 #include <map>
 #include <stdexcept>
@@ -447,6 +448,7 @@ class transport
             ++stats_.evicted_groups;
             return kind != 1 && deliver_packet(original);
         }
+        input_guard guard{*this, id};
         value.present[index] = true;
         ++value.count;
         std::vector<std::string> deliveries;
@@ -495,6 +497,7 @@ class transport
                 retained_bytes_ -= group_cost + value.bytes;
                 groups_.erase(it);
                 ++stats_.invalid_packets;
+                guard.release();
                 for (const auto& message : deliveries)
                 {
                     if (!deliver_packet(message))
@@ -517,6 +520,7 @@ class transport
             retained_bytes_ -= group_cost + value.bytes;
             groups_.erase(it);
         }
+        guard.release();
         // All group references are released before callbacks can stop the channel.
         for (const auto& message : deliveries)
         {
@@ -559,6 +563,31 @@ class transport
         std::array<bool, 32> delivered{};
     };
 
+    // Drop partially updated shards during unwinding; subsequent input may retry the group.
+    struct input_guard
+    {
+        transport& owner;
+        std::uint32_t id;
+        int exceptions = std::uncaught_exceptions();
+        bool active = true;
+
+        void release() noexcept {
+            active = false;
+        }
+
+        ~input_guard() {
+            if (active && std::uncaught_exceptions() > exceptions)
+            {
+                const auto it = owner.groups_.find(id);
+                if (it != owner.groups_.end())
+                {
+                    owner.retained_bytes_ -= group_cost + it->second.bytes;
+                    owner.groups_.erase(it);
+                }
+            }
+        }
+    };
+
     static constexpr std::size_t group_cost = sizeof(group) + 64;
 
     static std::string_view unpack(std::string_view bytes) {

tests/CMakeLists.txt

@@ -86,6 +86,7 @@ function(arknet_configure_allocation_failure_test target)
         BOOST_ASIO_DISABLE_SMALL_BLOCK_RECYCLING BOOST_ASIO_DISABLE_STD_ALIGNED_ALLOC BOOST_ASIO_DISABLE_BOOST_ALIGN
         ASIO_DISABLE_SMALL_BLOCK_RECYCLING ASIO_DISABLE_STD_ALIGNED_ALLOC ASIO_DISABLE_BOOST_ALIGN)
 endfunction()
+target_sources(arknet_fec_test PRIVATE scoped_allocation_failure.cpp)
 add_executable(arknet_allocation_failure_test allocation_failure.cpp scoped_allocation_failure.cpp test_main.cpp)
 arknet_configure_allocation_failure_test(arknet_allocation_failure_test)
 arknet_add_check(allocation_failure 30 arknet_allocation_failure_test)

tests/fec.cpp

@@ -4,13 +4,16 @@
 #include <cstdint>
 #include <limits>
 #include <memory>
+#include <new>
 #include <stdexcept>
 #include <string>
 #include <string_view>
 #include <vector>
 
+#include <arknet/arq/core.hpp>
 #include <arknet/fec/transport.hpp>
 #include "check.hpp"
+#include "scoped_allocation_failure.hpp"
 
 namespace {
 
@@ -101,6 +104,90 @@ DOCTEST_TEST_CASE("Rejected FEC metadata cannot advance the receive replay windo
     check(value.received == expected, "legitimate original still arrives");
 }
 
+DOCTEST_TEST_CASE("FEC input allocation failure releases incomplete recovery state and permits retry") {
+    for (const unsigned index : {1U, 2U})
+    {
+        DOCTEST_CAPTURE(index);
+        const auto settings = fixed(2, 1);
+        arknet::arq::options arq_settings;
+        arq_settings.mtu = 88;
+        arq_settings.congestion_control = false;
+        arknet::arq::control_block arq_sender(conversation, arq_settings), arq_receiver(conversation, arq_settings);
+        std::vector<std::string> wire;
+        unsigned forwarded = 0;
+        transport receiver(
+            conversation, mtu, settings, [](auto) {},
+            [&](auto bytes) {
+                ++forwarded;
+                return arq_receiver.input(bytes) == 0;
+            });
+        transport sender(
+            conversation, mtu, settings, [&](auto bytes) { wire.emplace_back(bytes); }, [](auto) { return true; });
+        arq_sender.set_output([&](auto bytes) { sender.send(bytes, 0); });
+        const std::vector<std::string> original{message(0), message(1)};
+        for (const auto& bytes : original)
+        {
+            check(arq_sender.send(bytes) == 0, "complete message queues in ARQ");
+        }
+        arq_sender.update(0);
+        check(wire.size() == 3, "two systematic ARQ datagrams and one parity shard are emitted");
+        check(receiver.input(wire[0], 0), "first systematic shard starts recovery group");
+        const auto before = scoped_allocation_failure::injections.load();
+        bool threw = false;
+        {
+            scoped_allocation_failure::scope failure;
+            try
+            { receiver.input(wire[index], 0); }
+            catch (const std::bad_alloc&)
+            { threw = true; }
+        }
+        check(threw && scoped_allocation_failure::injections == before + 1,
+            "systematic delivery staging or parity storage allocation fails once");
+        check(!scoped_allocation_failure::armed, "allocation failure is disarmed before checking recovery");
+        const auto failed = receiver.snapshot();
+        check(failed.retained_groups == 0 && failed.retained_bytes == 0,
+            "exception releases the incomplete group and its byte reservation");
+        check(receiver.input(wire[index], 1), "failed shard can be accepted again");
+        check(receiver.input(wire[index == 1 ? 0 : 1], 1), "an intact original completes the replacement group");
+        std::vector<std::string> received;
+        std::string bytes;
+        while (arq_receiver.receive(bytes))
+        {
+            received.push_back(bytes);
+        }
+        check(forwarded == 3, "the discarded group permits a previously forwarded original to reach ARQ again");
+        check(received == original, "ARQ delivers exact messages once and in order after the shard retry");
+        check(receiver.snapshot().retained_groups == 0 && receiver.snapshot().retained_bytes == 0,
+            "completed replacement group releases its cache");
+    }
+}
+
+DOCTEST_TEST_CASE("FEC staging cleanup preserves delivery callback exception semantics") {
+    fec_pair source(fixed(2, 1));
+    const auto original = source.send(2);
+    bool throw_delivery = true;
+    std::vector<std::string> received;
+    transport receiver(
+        conversation, mtu, fixed(2, 1), [](auto) {},
+        [&](auto bytes) {
+            if (throw_delivery)
+            {
+                throw std::bad_alloc();
+            }
+            received.emplace_back(bytes);
+            return true;
+        });
+    DOCTEST_CHECK_THROWS_AS(receiver.input(source.wire[0], 0), std::bad_alloc);
+    check(receiver.snapshot().retained_groups == 1 && receiver.snapshot().retained_bytes > 0,
+        "user callback exception preserves accepted recovery state");
+    throw_delivery = false;
+    check(!receiver.input(source.wire[0], 1), "accepted shard stays duplicate after its delivery callback throws");
+    check(receiver.input(source.wire[1], 1), "later original completes the accepted recovery group");
+    check(received == std::vector<std::string>{original[1]}, "callback exception does not schedule another delivery");
+    check(receiver.snapshot().retained_groups == 0 && receiver.snapshot().retained_bytes == 0,
+        "completed group releases the accepted state");
+}
+
 DOCTEST_TEST_CASE("Adaptive FEC compares segment and datagram loss ratios independently") {
     auto settings = arknet::fec::options::adaptive();
     transport sender(conversation, mtu, settings, [](auto) {}, [](auto) { return true; });