fix(redis): reject malformed RESP3 reply boundaries

docs/features/redis/configuration.md

@@ -18,7 +18,9 @@ Configure `redis::connect_options` before `async_connect`: `host`, `port`, `user
 | `map`, `attribute` | Alternating key/value entries in `elements`, retaining all key types |
 | `null` | No payload |
 
-`attributes` contains RESP3 attributes attached to the value. Inspect types before accessing data. A server error is a typed reply, not a transport failure: `bool(result)` checks `client_errc`, so applications must also inspect individual replies. Failed pipelines can contain replies received before the failure.
+`attributes` contains RESP3 attributes attached to the value. Inspect types before accessing data. A server error is a typed reply, not a transport failure: `bool(result)` checks `client_errc`, so applications must also inspect individual replies. Failed or cancelled pipelines retain complete replies received before completion; those values remain valid after subsequent decoding and client destruction. Partial replies are never exposed.
+
+The decoder follows the [Redis protocol specification](https://redis.io/docs/latest/develop/reference/protocol-spec/): RESP3 double fractions require digits after the decimal point, and pushes are top-level messages. A malformed reply or a `response_limits` violation closes the connection with `client_errc::protocol_error` and fails outstanding requests without replay.
 
 ## Limits and deadlines
 

docs/features/redis/configuration_CN.md

@@ -18,7 +18,9 @@
 | `map`、`attribute` | `elements` 中交替排列键和值，保留键的类型 |
 | `null` | 无数据 |
 
-`attributes` 保存附着在该值上的 RESP3 属性。读取数据前先检查类型。服务端错误是类型化回复，不是传输失败：`bool(result)` 检查 `client_errc`，应用仍需检查每一项回复。流水线失败时，结果可能保留失败前已经收到的回复。
+`attributes` 保存附着在该值上的 RESP3 属性。读取数据前先检查类型。服务端错误是类型化回复，不是传输失败：`bool(result)` 检查 `client_errc`，应用仍需检查每一项回复。流水线失败或取消时，结果保留完成前已经收到的完整回复；后续解码和客户端析构不会使这些值失效。不完整的回复不会暴露给调用者。
+
+解码器遵循 [Redis 协议规范](https://redis.io/docs/latest/develop/reference/protocol-spec/)：RESP3 double 的小数点后必须有数字，推送必须位于消息顶层。回复格式错误或超出 `response_limits` 时，以 `client_errc::protocol_error` 关闭连接，让在途请求失败，不重放命令。
 
 ## 限额与超时
 

include/arknet/redis/detail/resp.hpp

@@ -139,6 +139,10 @@ class resp_decoder
                 {
                     return fail(resp_status::malformed);
                 }
+                else if (byte == '>' && !frames_.empty())
+                {
+                    return fail(resp_status::malformed);
+                }
                 else if (++values_ > limits_.max_values || frames_.size() >= limits_.max_depth)
                 {
                     return fail(resp_status::limit_exceeded);
@@ -563,6 +567,12 @@ class resp_decoder
             {
                 return fail(resp_status::malformed);
             }
+            const auto point = integral.find('.');
+            if (point != std::string_view::npos &&
+                (point + 1 == integral.size() || integral[point + 1] < '0' || integral[point + 1] > '9'))
+            {
+                return fail(resp_status::malformed);
+            }
             const auto parsed = std::from_chars(number.data(), number.data() + number.size(), value.real);
             if (parsed.ec != std::errc{} || parsed.ptr != number.data() + number.size() || !std::isfinite(value.real))
             {

tests/redis_client.cpp

@@ -269,6 +269,87 @@ DOCTEST_TEST_CASE("Redis disconnect and malformed replies fail outstanding comma
     }
 }
 
+DOCTEST_TEST_CASE("Redis failed pipelines retain complete owning replies and retire completion captures") {
+    const std::string binary("a\0\r\nb", 5);
+    const std::vector<std::pair<std::string, client_errc>> failures{{",1.\r\n+VALID\r\n", client_errc::protocol_error},
+        {"*1\r\n>1\r\n+invalid\r\n+VALID\r\n", client_errc::protocol_error}, {"$3\r\nab", client_errc::disconnected}};
+    for (const auto& [wire, error] : failures)
+    {
+        server endpoint([&](peer& remote) {
+            remote.hello();
+            remote.expect({"ECHO", binary});
+            remote.expect({"GET", "broken"});
+            remote.expect({"PING"});
+            remote.send("$5\r\n" + binary + "\r\n" + wire);
+        });
+        auto connection = std::make_unique<client>();
+        check(connect(*connection, endpoint.port()), "connected");
+        auto capture = std::make_shared<int>(1);
+        const std::weak_ptr<int> retained = capture;
+        std::atomic<unsigned> calls{};
+        std::promise<result> failed, following;
+        auto failed_future = failed.get_future(), following_future = following.get_future();
+        connection->async_pipeline({{"ECHO", binary}, {"GET", "broken"}}, [&, capture](result reply) {
+            ++calls;
+            failed.set_value(std::move(reply));
+        });
+        capture.reset();
+        connection->async_execute({"PING"}, [&](result reply) { following.set_value(std::move(reply)); });
+        auto response = await(failed_future);
+        check(response.error == error && response.command_may_have_run, "failure preserves unknown write outcome");
+        check(await(following_future).error == error, "later command fails without receiving a stray reply");
+        check(connection->close_and_wait(), "retire failed connection");
+        connection.reset();
+        endpoint.finish();
+        check(calls == 1 && retained.expired(), "complete once and release callback captures");
+        check(response.replies.size() == 1 && response.replies[0].type == value_type::bulk_string &&
+                  response.replies[0].text == binary,
+            "complete prefix survives parser and client destruction");
+    }
+}
+
+DOCTEST_TEST_CASE("Redis cancellation preserves an owning pipeline prefix while later replies drain") {
+    const std::string binary("a\0\r\nb", 5);
+    std::promise<void> release;
+    auto allowed = release.get_future();
+    server endpoint([&](peer& remote) {
+        remote.hello();
+        remote.expect({"ECHO", binary});
+        remote.expect({"INCR", "cancelled"});
+        remote.expect({"GET", "following"});
+        remote.send("$5\r\n" + binary + "\r\n>1\r\n+prefix-ready\r\n");
+        if (allowed.wait_for(5s) != std::future_status::ready)
+        {
+            throw std::runtime_error("Missing drain signal");
+        }
+        remote.send(":22\r\n:33\r\n");
+    });
+    client connection;
+    std::promise<void> prefix;
+    auto prefix_ready = prefix.get_future();
+    connection.set_push_handler([&](auto) { prefix.set_value(); });
+    check(connect(connection, endpoint.port()), "connected");
+    std::atomic<unsigned> calls{};
+    std::promise<result> cancelled, following;
+    auto cancelled_future = cancelled.get_future(), following_future = following.get_future();
+    auto id = connection.async_pipeline({{"ECHO", binary}, {"INCR", "cancelled"}}, [&](result reply) {
+        ++calls;
+        cancelled.set_value(std::move(reply));
+    });
+    connection.async_execute({"GET", "following"}, [&](result reply) { following.set_value(std::move(reply)); });
+    await(prefix_ready);
+    connection.cancel(id);
+    auto response = await(cancelled_future);
+    check(response.error == client_errc::cancelled && response.command_may_have_run, "cancel sent pipeline");
+    release.set_value();
+    auto next = await(following_future);
+    check(next && next.replies.size() == 1 && next.replies[0].integer == 33, "remaining reply keeps its old slot");
+    check(connection.close_and_wait(), "retire drained connection");
+    endpoint.finish();
+    check(calls == 1 && response.replies.size() == 1 && response.replies[0].text == binary,
+        "cancelled owning prefix survives subsequent decoding and shutdown");
+}
+
 DOCTEST_TEST_CASE("Redis owner destruction inside a completion aborts later slots safely") {
     server endpoint([](peer& remote) {
         remote.hello();

tests/redis_resp.cpp

@@ -90,6 +90,46 @@ DOCTEST_TEST_CASE("RESP3 represents scalars, maps, sets, pushes and verbatim val
     check(push.type == resp_type::push && push.elements[0].text == "invalidate", "push remains distinguishable");
 }
 
+DOCTEST_TEST_CASE("RESP3 doubles require digits after a decimal point") {
+    check(decode(",1.0\r\n").real == 1.0, "fractional digits");
+    check(decode(",-1.25e+2\r\n").real == -125.0, "fraction and exponent");
+    check(decode(",+1e-2\r\n").real == 0.01, "exponent without a fraction");
+    for (const std::string wire : {",1.\r\n", ",-1.e+2\r\n", ",+0.E-1\r\n"})
+    {
+        DOCTEST_INFO(wire);
+        resp_decoder decoder;
+        resp_value value;
+        value.text = "unchanged";
+        check(decoder.append(wire + "+VALID\r\n"), "append invalid double and following reply");
+        check(decoder.next(value) == resp_status::malformed && value.text == "unchanged",
+            "incomplete fraction is not a reply");
+        check(decoder.next(value) == resp_status::malformed, "invalid double cannot expose the following reply");
+    }
+}
+
+DOCTEST_TEST_CASE("RESP3 pushes cannot be nested inside another value") {
+    for (const std::string aggregate : {"*1", "%1", "~1", "|1", ">2\r\n+outer", "*?", "%?", "~?"})
+    {
+        DOCTEST_INFO(aggregate);
+        const auto wire = aggregate + "\r\n>1\r\n+invalid\r\n:1\r\n+VALID\r\n";
+        for (std::size_t split = 0; split < wire.size(); ++split)
+        {
+            resp_decoder decoder;
+            resp_value value;
+            value.text = "unchanged";
+            check(decoder.append(std::string_view(wire).substr(0, split)), "append prefix");
+            auto status = decoder.next(value);
+            if (status == resp_status::need_more)
+            {
+                check(decoder.append(std::string_view(wire).substr(split)), "append suffix");
+                status = decoder.next(value);
+            }
+            check(status == resp_status::malformed && value.text == "unchanged", "reject nested push");
+            check(decoder.next(value) == resp_status::malformed, "nested push failure is sticky");
+        }
+    }
+}
+
 DOCTEST_TEST_CASE("RESP3 attributes decorate the correct nested value without consuming a reply") {
     auto value = decode("|1\r\n+outer\r\n:1\r\n|1\r\n+other\r\n:2\r\n"
                         "*2\r\n|1\r\n+ttl\r\n:9\r\n$1\r\na\r\n:5\r\n");