diff --git a/docs/features/rudp/fec.md b/docs/features/rudp/fec.md index 5f3dda0..b0115e3 100644 --- a/docs/features/rudp/fec.md +++ b/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 | diff --git a/docs/features/rudp/fec_CN.md b/docs/features/rudp/fec_CN.md index b5e6b90..49887a7 100644 --- a/docs/features/rudp/fec_CN.md +++ b/docs/features/rudp/fec_CN.md @@ -69,6 +69,7 @@ client.set_fec_options(settings); | 突发丢失超过 2 个分片 | 仍需 ARQ | | 实际校验数 | 字节预算与最小分组大小可能使其低于配置值 | | 组保留期限 | 完整组立即释放，未完成组到期释放 | +| 分片暂存分配失败 | `fec::transport::input` 释放受影响的未完成恢复组并继续抛出异常；应用交付回调抛异常时保留已接收状态 | | 缓存不足 | 跳过额外 FEC 保留；合法原始数据继续进入 ARQ，可靠交付仍生效 | | 重复过滤 | 固定 4096 组窗口 | | 编解码临时空间 | 受分片数与 MTU 限制，在保留预算之外另计 | diff --git a/include/arknet/fec/transport.hpp b/include/arknet/fec/transport.hpp index 0635c5d..943cde4 100644 --- a/include/arknet/fec/transport.hpp +++ b/include/arknet/fec/transport.hpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -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 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 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) { diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index beaa0f4..f0b75b1 100644 --- a/tests/CMakeLists.txt +++ b/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) diff --git a/tests/fec.cpp b/tests/fec.cpp index 6068c17..4911e6c 100644 --- a/tests/fec.cpp +++ b/tests/fec.cpp @@ -4,13 +4,16 @@ #include #include #include +#include #include #include #include #include +#include #include #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 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 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 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 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{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; });