fix(security): bound RESP processing and pipeline queues

Constraint: Bound unauthenticated RESP parsing and optional pipeline resources without changing Redis command, storage, or Raft semantics.

Confidence: High. Parser budgets, every active network consumer, legacy queue draining, and bounded pipeline admission have direct regression coverage.

Scope-risk: Oversized or excessively expensive RESP frames now fail closed; queue admission shares the existing 30-second timeout.

Red: Baseline parsing trusted attacker-controlled lengths, replayed incomplete aggregates without a cumulative budget, retained legacy command copies, and used an unbounded pipeline channel.

Green: Resource-limit, parser-consumer, pipeline, and Linux TCP isolation regressions pass.

Tested: Windows MSVC cargo test -p resp --all-features --locked; cargo test -p net --lib --all-features --locked; strict resp/net Clippy; fmt check; diff check. WSL/Linux exact oversized unauthenticated request and protocol-error isolation tests.

Not-tested: Full WSL workspace suite. The earlier Windows workspace run reached all unit suites but retained the repository baseline of 19 TCP startup failures.

Co-authored-by: OmX <omx@oh-my-codex.dev>

Author: AlexStocks

.planning/KANBAN.md

@@ -14,7 +14,7 @@
 
 | ID | 工作项 | Requirement / Gate 关系 | 当前状态 |
 |---|---|---|---|
-| `RESP-LIMITS-001` | 限制未认证 RESP Array/Map/Set/Push 声明长度、初始预分配和嵌套深度 | `REQ-COMPAT-002`、`REQ-COMPAT-006`、`REQ-WORK-003` | PR `#404` 已创建；独立 review 缺口已修复，二次 review 无 finding，最终 Head/checks 以 GitHub 为准 |
+| `RESP-LIMITS-001` | 有界 RESP 解析、未认证连接 buffer、parser 历史副本和 optional pipeline 背压 | `REQ-COMPAT-002`、`REQ-COMPAT-006`、`REQ-STABILITY-002`、`REQ-STABILITY-003`、`REQ-WORK-003` | PR `#404` 已创建；本地 Windows/WSL 定向验证完成，待发布并复检最终 Head CI |
 
 ## Ready for a separate implementation task
 

.planning/STATE.md

@@ -8,29 +8,29 @@
 >
 > 实现基线：`main` at `cbc28958f261ae049d67a8b4a9d904d794b37726`
 >
-> 状态：PR 已创建；独立 review 的嵌套深度缺口已修复，二次 review 无 finding，最终 Head/checks 以 GitHub 实时查询为准
+> 状态：PR 已创建；嵌套深度及广义资源边界缺口已在本地修复并验证，待发布后复检最终 Head/checks
 >
-> 当前范围：拒绝超出 Redis 8.8.1 整数边界的 RESP 聚合长度，限制 Array/Map/Set/Push 的初始预分配和递归嵌套深度
+> 当前范围：限制 RESP 首行、payload、buffer、递归、解码节点和重复解析工作；限制未认证连接 buffer；清理 parser 历史副本；为 optional pipeline 建立真实背压
 >
 > Requirement 边界：`REQ-COMPAT-002`、`REQ-COMPAT-006`、`REQ-WORK-003`
 
 ## 当前目标
 
-本 task 修复 Issue #395 B1 中已经由源码确认的未认证 RESP 聚合类型无界预分配问题。客户端声明长度不得直接控制 `Vec` 的初始容量；超出 Redis 8.8.1 `INT_MAX` 边界的声明返回协议错误，合法声明的初始容量最多为 1024，聚合递归最多为 128 层。
+本 task 修复 Issue #395 B1 中已经由源码确认的未认证 RESP 资源耗尽路径，并处理 Issue #398 的 optional pipeline 无界 channel。客户端声明长度不得直接控制分配；解析器和未认证连接具有内存、深度、对象数与累计工作预算；`command_queue_size` 形成真实背压。
 
-本 task 不处理实际流入的超大 bulk 或连接累计 buffer 限额，不修改 PR #402 的文档，不处理 Issue #395 的其他条目，也不实现 Redis Oracle provenance 或 Embedded Redis Hot Tier。
+经调用链复核，Issue #395 的 `DEL`、`MSETNX` 和 expiration P0 描述与当前 executor gate、单写者 Raft apply 及 no-op compaction 不符。本 PR 不添加可能与 Raft apply 死锁的无效 record lock，也不修改 PR #402 的文档、Redis Oracle provenance 或 Embedded Redis Hot Tier。
 
 ## 当前授权边界
 
 允许：
 
-- 修改 `src/resp/src/parse.rs`、本 task 的设计/计划、`.planning/STATE.md` 和 `.planning/KANBAN.md`。
-- 运行 resp crate 单测、Clippy、格式检查、Git diff 检查及与 changed surface 对应的验证。
-- commit、push 并创建以 `main` 为 base 的独立 PR；发布后实时查询 checks 和 review threads。
+- 修改 `src/resp/src/parse.rs`、`src/net` 下 parser consumer、pipeline 和定向测试、本 task 的设计/计划、`.planning/STATE.md` 和 `.planning/KANBAN.md`。
+- 运行 resp/net 定向测试、workspace 测试、Clippy、格式检查、Git diff 检查和 WSL/Linux TCP 验证。
+- commit 并 fast-forward push 到现有 PR #404 的 head 分支；发布后实时查询 checks 和 review threads。
 
 禁止：
 
-- 修改 Cargo、网络/认证逻辑、其他生产源码、PR #402 文档、Oracle provenance 或 Hot Tier 实现。
+- 修改 Cargo、storage/Raft/命令语义、PR #402 文档、Oracle provenance 或 Hot Tier 实现。
 - 在旧 `redis-8.8.1-stability-foundation` worktree 继续、暂存、提交、push、清理或回退六文件实现草稿。
 - 把旧草稿的绿色测试、审查或真实构建准备表述为方案 A 已实现。
 - 扩大到 Embedded Redis Hot Tier、Redis fork、动态库、loader、Cache ON 或组合发行实现。
@@ -41,6 +41,9 @@
 - PR `#383` 已于 2026-07-28 合并：final Head `42c16bef899385bd2e1b1e16e2e0202d4a614590`，merge commit `58030e1331655546ea4547a9a94efc493534ef7d`；它只完成 Oracle 方案 A 的规划闭环。
 - PR `#388` 已于 2026-07-30 合并；本 task 是从最新 `main` 创建的独立 implementation task。
 - RESP 聚合长度上限采用 Redis 8.8.1 的 `INT_MAX`，初始预分配上限采用 1024。
+- 所有 RESP 首行上限为 64 KiB，bulk payload 上限为 512 MiB，通用 parser buffer 上限为 1 GiB，未认证连接 buffer 上限为 1 MiB。
+- 单 frame 最多物化 65,536 个 RESP 节点；分片 aggregate 重放最多累计 1,000,000 次节点访问。
+- optional pipeline 使用容量至少为 1 的 bounded Tokio channel，queue admission 纳入现有 30 秒 timeout。
 - `D011`：Redis Oracle required provenance 采用 verifier fresh-checkout independent rebuild 和 exact binary hash equality。
 - `D012`：规划 task 与实施 task 分离；规划批准不授权源码实现，提前产生的实现草稿冻结。
 - Redis 8.8.1 tag `8.8.1` / commit `77b6c308396c9700672390a210143a8496fb4b10` 是唯一兼容和 Oracle 基线。
@@ -114,13 +117,18 @@ D:\test\github\kiwi\.worktrees\redis-8.8.1-stability-foundation\.codex\recovery\
 - 最终工作区 Windows 与 WSL `cargo clippy -p resp --all-targets -- -D warnings -D clippy::unwrap_used`：通过。
 - 最终工作区 `cargo fmt --all -- --check` 与 `git diff --check`：通过。
 - 二次独立 review 未发现 Critical、Important 或 Minor；确认 128/129 深度边界、四类聚合错误传播和容量 helper 测试与设计一致。
+- 后续广义审计新增 3 个 Important：非 inline 首行绕过、分片 aggregate O(N²) 重放、协议 count 与实际对象预算混淆；均已通过统一首行、65,536 节点和 1,000,000 node-visits 预算修复。
+- 最终 consolidation worktree Windows `cargo test -p resp --all-features --locked`：71 unit + 20 integration + doc tests 全部通过。
+- 最终 consolidation worktree Windows `cargo test -p net --lib --all-features --locked`：32/32 通过；严格 resp/net Clippy 以 `-D warnings -D clippy::unwrap_used` 通过。
+- WSL/Linux RESP 结果同为 71 + 20 + doc tests；新增未认证超限 TCP 与既有 protocol-error 控制用例均为 1/1 通过。
+- Windows workspace 已运行；到达的 unit suites 全部通过，19 个 TCP 用例统一复现仓库基线 `server did not become connectable`，不作为本分支回归。
 - checks、review threads 和 PR 状态不在本文件中缓存；任何当前结论必须重新查询 GitHub。
 
 PR `#383` 的结果只证明 Oracle 规划闭环，不证明方案 A 已实现；PR `#388` 也不改变该结论。
 
 ## 下一条安全动作
 
-1. 提交并 push PR #404 review 修复后，重新查询最终 Head 的 checks、评论和 review threads；不得把旧 Head 的 CI 结果作为最终状态。
+1. 提交并 fast-forward push PR #404 的 consolidation 修复后，重新查询最终 Head 的 checks、评论和 review threads；不得把旧 Head 的 CI 结果作为最终状态。
 2. 若最终 Head checks 未完成，只报告 pending，不给可 Merge 结论。
 3. 不 Resolve 或回复 #402/#404 review thread，不 merge PR。
 4. PR `#383` 的规划历史保持不变，旧六文件 Oracle 草稿继续冻结。

docs/superpowers/plans/2026-07-31-bounded-request-processing.md

@@ -0,0 +1,63 @@
+# Bounded Request Processing Implementation Plan
+
+> **For AI workers:** Required sub-skill: use superpowers:subagent-driven-development to implement this plan task by task. Track steps with checkboxes.
+
+**Goal:** Bound unauthenticated RESP parsing and optional pipeline queuing, and stop active connections from retaining duplicate command history.
+
+**Architecture:** Keep the existing parser and network APIs, add explicit parser limits with depth-aware recursive parsing, enforce the unauthenticated limit at the active network boundary, drain the legacy command queue, and replace the optional pipeline's unbounded channel with a bounded sender. Each behavior is introduced through a failing regression test before production code changes.
+
+**Tech Stack:** Rust 1.97.1, nom 8, Tokio mpsc, Cargo tests, Windows MSVC and WSL/Linux validation.
+
+---
+
+### Task 1: Bound RESP lengths, allocation, buffering, and nesting
+
+**Files:**
+- Modify: `src/resp/src/parse.rs`
+- Modify: `src/resp/src/error.rs` only if a structured limit variant is required
+
+- [x] Add parser tests using small private limits for bulk length, aggregate length, exact boundaries, nesting depth, chunked buffer growth, and reset-after-error.
+- [x] Run `cargo +1.97.1-x86_64-pc-windows-msvc test -p resp --lib parse::tests --locked` and confirm the new assertions fail because over-limit frames are incomplete or complete.
+- [x] Add immutable parser limits, pre-append buffer checks, depth-aware aggregate parsing, bounded initial capacity, and fallible incremental reservation.
+- [x] Add uniform first-line limits, a decoded-node budget, and a cumulative incomplete-frame parse-work budget after code-quality review.
+- [x] Re-run the parser tests and all `resp` tests; require zero failures.
+- [x] Commit `fix(resp): bound request parsing resources`.
+
+### Task 2: Enforce pre-authentication limits and drain parser history
+
+**Files:**
+- Modify: `src/net/src/network_handle.rs`
+- Modify: `src/net/src/optimized_handler.rs`
+- Modify: `src/net/tests/storage_command_e2e_tests.rs`
+- Modify: `src/resp/src/parse.rs` only for read-only buffer/queue observability
+
+- [x] Add a TCP regression test that sends an incomplete request beyond 1 MiB before `AUTH`, observes connection closure, then authenticates and pings through a healthy connection.
+- [x] Add a parser/network regression proving complete frames do not accumulate undrained legacy commands in active consumers.
+- [x] Run the two new tests and confirm timeout or retained-command failures on the baseline implementation.
+- [x] Check `buffered_len + read_len` before unauthenticated parsing and drain one legacy command result after every complete frame in all consumers.
+- [x] Re-run the new tests plus the existing bad-connection protocol test; require zero failures under WSL/Linux.
+- [x] Commit `fix(net): cap unauthenticated request buffering`.
+
+### Task 3: Enforce bounded pipeline admission
+
+**Files:**
+- Modify: `src/net/src/pipeline.rs`
+
+- [x] Add tests proving zero capacity normalizes to one and that the production channel helper rejects immediate admission after its configured capacity is occupied.
+- [x] Run `cargo +1.97.1-x86_64-pc-windows-msvc test -p net pipeline::tests --locked` and confirm the bounded-capacity assertions fail.
+- [x] Replace `UnboundedSender/UnboundedReceiver` with bounded Tokio mpsc, await admission under the existing timeout, and report actual normalized capacity.
+- [x] Re-run pipeline and net tests; require zero failures.
+- [x] Commit `fix(net): enforce pipeline queue capacity`.
+
+### Task 4: Integrate, review, and publish one PR
+
+**Files:**
+- Modify: this plan's checkboxes as each gate completes
+
+- [x] Run `cargo fmt --all -- --check`.
+- [x] Run strict Clippy for `resp` and `net` with all targets/features and `-D warnings -D clippy::unwrap_used`.
+- [x] Run `cargo test --workspace --all-features --locked` with `RUST_TEST_THREADS=1`; all reached unit suites pass, while the 19 Windows TCP tests reproduce the repository baseline inability to start the server.
+- [x] Run the targeted RESP and network tests under WSL/Linux with an independent target directory.
+- [x] Run `git diff --check` and verify only planned files changed.
+- [x] Complete specification compliance review, then code-quality review, and resolve every important finding.
+- [ ] Fast-forward the existing PR #404 head `codex/fix-resp-parser-limits` with the consolidated fix; do not merge it.

docs/superpowers/plans/2026-07-31-resp-parser-resource-limits.md

@@ -1,5 +1,8 @@
 # RESP 聚合类型资源限额实现计划
 
+> 最终 PR 范围已扩展，本文件作为聚合预分配与 nesting 阶段计划保留；现行计划见
+> `2026-07-31-bounded-request-processing.md`。
+
 > **面向 AI 代理的工作者：** 必需子技能：使用 superpowers:subagent-driven-development（推荐）或 superpowers:executing-plans 逐任务实现此计划。步骤使用复选框（`- [ ]`）语法来跟踪进度。
 
 **目标：** 阻止未认证客户端通过 RESP 聚合类型声明长度或递归嵌套触发无界初始分配、capacity-overflow panic 或栈耗尽。

docs/superpowers/specs/2026-07-31-bounded-request-processing-design.md

@@ -0,0 +1,90 @@
+# Bounded Request Processing Design
+
+## Goal
+
+Prevent unauthenticated or slow clients from causing unbounded memory growth in
+the active RESP parsing path and the optional command pipeline, while preserving
+Redis-compatible request limits and existing command behavior.
+
+## Scope
+
+The change covers three proven ownership problems:
+
+1. RESP length headers can trigger oversized aggregate reservations before
+   authentication, bulk frames can grow the parser buffer without a declared
+   length check, and recursive aggregates have no nesting limit.
+2. The active network path returns parsed `RespData` but leaves a second
+   `RespCommand` copy in `RespParse`, retaining command history for the lifetime
+   of a connection.
+3. `PipelineConfig::command_queue_size` is reported but ignored because the
+   optional pipeline uses an unbounded channel.
+
+The storage findings from issue #395 are not implementation scope. Normal
+client `DEL` and `MSETNX` calls already hold the executor's
+`STORAGE_EXCLUSIVE` gate, Raft apply is single-writer, and expiration compaction
+is currently a no-op. Those facts invalidate their current P0 data-loss
+descriptions; adding local locks would not repair the claimed Raft interleaving.
+
+## RESP Limits
+
+`RespParse` owns immutable limits. Production constructors use:
+
+- 64 KiB maximum first-line length for inline commands and every RESP type or
+  aggregate header.
+- 512 MiB maximum bulk, bulk-error, and verbatim-string payload length.
+- 1 GiB maximum buffered frame length for authenticated/general parsing.
+- `i32::MAX` maximum aggregate item or pair count, matching Redis' multibulk
+  header boundary.
+- 128 maximum aggregate nesting depth.
+- At most 1024 elements of initial aggregate reservation, followed by fallible
+  incremental reservation.
+- At most 65,536 decoded `RespData` nodes in one frame. The `i32::MAX` wire
+  count remains accepted as a protocol boundary, but it cannot force that many
+  objects to be materialized.
+- At most 1,000,000 cumulative node visits while an incomplete frame is
+  reparsed. This bounds repeated aggregate-prefix work without changing the
+  public parser API.
+
+Tests use a private constructor with small limits, so boundaries are exercised
+without allocating large buffers. Limits are checked before extending the
+buffer or reserving aggregate memory. Limit violations are terminal parser
+errors rather than incomplete frames.
+
+The active network path also enforces a 1 MiB buffer limit before authentication.
+It checks the existing incomplete-frame size plus the next read before passing
+bytes to the parser. A violation closes only that connection through the
+existing protocol-error path.
+
+## Parser Ownership
+
+The public parser API remains compatible in this PR. All network-side parser
+consumers route through one helper that drains the corresponding legacy
+`next_command()` entry after each complete frame. This keeps the internal queue
+bounded without combining an API removal with the security fix. Removing the
+duplicate command representation is a follow-up API cleanup.
+
+## Pipeline Backpressure
+
+The optional pipeline uses `tokio::sync::mpsc::channel` with a minimum capacity
+of one. `submit_command` awaits bounded-channel capacity and applies the existing
+30-second timeout to queue admission as well as response delivery. Channel
+closure remains a distinct error. Statistics report the normalized, real queue
+capacity.
+
+## Error Handling
+
+- Malformed or over-limit RESP frames become `RespParseResult::Error`.
+- An over-limit unauthenticated connection is closed without affecting other
+  clients or the server process.
+- Queue admission timeout returns `PipelineError::Timeout`.
+- No production path uses `unwrap`, `expect`, or infallible allocation for
+  attacker-controlled aggregate capacity.
+
+## Verification
+
+The implementation must prove red-green behavior for oversized first lines and
+length headers, nesting, decoded-node amplification, cumulative incomplete
+aggregate work, chunked buffer growth, parser reset, legacy queue draining, and
+bounded pipeline capacity. Final gates are targeted tests, workspace tests,
+strict Clippy, formatting, `git diff --check`, Linux/WSL verification, and
+current PR Head/check reconciliation.

docs/superpowers/specs/2026-07-31-resp-parser-resource-limits-design.md

@@ -1,5 +1,8 @@
 # RESP 聚合类型资源限额设计
 
+> 最终 PR 范围已扩展，本文件作为聚合预分配与 nesting 阶段记录保留；现行合同见
+> `2026-07-31-bounded-request-processing-design.md`。
+
 > 状态：已批准
 > 日期：2026-07-31
 > 基线：`main` at `cbc28958f261ae049d67a8b4a9d904d794b37726`

src/net/src/async_resp_parser.rs

@@ -72,7 +72,10 @@ impl AsyncRespParser {
 
         // Parse as many complete commands as possible
         loop {
-            let parse_result = self.parser.parse(self.buffer.split().freeze());
+            let parse_result = crate::network_handle::parse_request(
+                &mut self.parser,
+                self.buffer.split().freeze(),
+            );
 
             match parse_result {
                 RespParseResult::Complete(command_data) => {
@@ -230,6 +233,18 @@ mod tests {
         assert_eq!(parser.buffered_bytes(), 0);
     }
 
+    #[test]
+    fn parse_data_discards_underlying_legacy_command_copy() {
+        let mut parser = AsyncRespParser::new(RespVersion::RESP2);
+
+        let commands = parser
+            .parse_data(Bytes::from_static(b"*1\r\n$4\r\nPING\r\n"))
+            .expect("valid command should parse");
+
+        assert_eq!(commands.len(), 1);
+        assert!(parser.parser.next_command().is_none());
+    }
+
     #[test]
     fn test_command_batch_priority() {
         let ping_cmd = RespData::Array(Some(vec![RespData::BulkString(Some(Bytes::from("PING")))]));

src/net/src/handle.rs

@@ -24,11 +24,12 @@ use cmd::table::CmdTable;
 use executor::{CmdExecution, CmdExecutor};
 use log::error;
 use resp::encode::RespEncoder;
-use resp::{Parse, RespData, RespEncode, RespParseResult};
+use resp::{RespData, RespEncode, RespParseResult};
 use storage::storage::Storage;
 use tokio::select;
 use tokio_util::sync::CancellationToken;
 
+use crate::network_handle::parse_client_request;
 use crate::storage_client::StorageClient;
 
 pub async fn process_connection(
@@ -47,8 +48,11 @@ pub async fn process_connection(
                     Ok(n) => {
                         if n == 0 { return Ok(()); }
 
-                        let parse_result =
-                            resp_parser.parse(Bytes::copy_from_slice(&buf[..n]));
+                        let parse_result = parse_client_request(
+                            &mut resp_parser,
+                            client.is_authenticated(),
+                            Bytes::copy_from_slice(&buf[..n]),
+                        )?;
                         match parse_result {
                             RespParseResult::Complete(data) => {
                                 if let RespData::Array(Some(params)) = data {