fix(xhttp): rewrite libx bench servers on current API + h2 on_data arg fix

issues/http-h2-ondata-user-arg.md

@@ -0,0 +1,63 @@
+# Issue: HTTP/2 下 on_data 回调 arg 与 h1 不一致 —— per-request 状态不可见
+
+> 状态：fixed（`proto_h2.c` on_data 的 arg 对齐 h1 的 `stream->user ?: route_info->arg`）
+> 首次观测：2026-08-26，重写 `libx/bench/http/https_bench_server.cpp` 到新 server API
+> 后用 curl 冒烟测试 POST 时
+
+## 现象
+
+HTTPS bench server 的 `POST /echo` 导致进程崩溃（curl `exit 56` connection reset）：
+
+- `curl --http1.1 POST` → 正常返回 echo
+- `curl POST`（默认 ALPN 协商到 **h2**）→ server 崩溃
+- `curl -v` 确认：`ALPN: server accepted h2`
+
+## 根因
+
+h1 与 h2 的 `on_data` 回调 arg 解析**不一致**：
+
+- `proto_h1.c` `on_body`（:113）：
+
+```c
+  void *arg = stream->user ? stream->user : stream->route_info->arg;
+```
+
+  优先用 per-request 的 `stream->user`（由 `xHttpCtxSetUser` 设置）。
+
+- `proto_h2.c` `h2_on_data_chunk_recv_callback`（:148，修复前）：
+
+```c
+  int rc = stream->route_info->on_data((const char *)data, len, stream->route_info->arg);
+```
+
+  **直接传 route 级 `route_info->arg`**，完全忽略 `stream->user`。
+
+标准用法（`on_request` 里 `xHttpCtxSetUser(ctx, per_request_state)`，`on_data` 里取回）
+在 h1 下成立，在 h2 下 `on_data` 拿到的是 route 级 arg（通常是 NULL）——bench 的
+`echo_on_data` 解引用 NULL → crash。任何依赖 per-request user data 的 h2 handler
+都会踩中。`on_done` 无此问题（`conn_dispatch_request` 在 server.c，两个协议共享，
+用 `stream->user ?: route_info->arg`）。
+
+## 修复
+
+`proto_h2.c` 的 `h2_on_data_chunk_recv_callback` 改为与 h1 相同的 arg 解析：
+
+```c
+void *arg = stream->user ? stream->user : stream->route_info->arg;
+int   rc  = stream->route_info->on_data((const char *)data, len, arg);
+```
+
+## 验证
+
+- 修复后 `curl -sk -X POST -d body https://127.0.0.1:8443/echo`（h2）正常 echo；
+  `--http1.1` 也正常。
+- 现有测试：h2 测试（`server_h2_test.cpp`）通过（原测试的 on_data 用 route 级
+  arg 静态状态，未覆盖 per-request user data 路径，所以此前未暴露）。
+
+## 复现
+
+```bash
+# 构建 https_bench_server（X_BUILD_BENCHMARKS=ON）并启动
+# POST 用默认协商（h2）即崩
+curl -sk -X POST -d hi https://127.0.0.1:8443/echo
+```

libx/bench/http/http_bench_server.cpp

@@ -13,7 +13,8 @@
  *   GET  /echo?size=N   → 200 with N bytes of 'x' (payload test)
  *   POST /echo          → 200 echoing the request body
  *
- * Designed to be benchmarked with wrk, ab, or hey:
+ * Built on the current xhttp server API (xHttpMux resolver + xHttpCtx
+ * handlers). Designed to be benchmarked with wrk, ab, or hey:
  *   wrk -t4 -c100 -d10s http://127.0.0.1:8080/ping
  *   hey -n 100000 -c 50 http://127.0.0.1:8080/ping
  */
@@ -29,41 +30,79 @@
 
 static xEventLoop g_loop = nullptr;
 
-// GET /ping → "pong"
-static void handle_ping(xHttpResponseWriter writer, const xHttpRequestConf *req, void *arg) {
-  (void)req;
+/* GET /ping → "pong" */
+static void handle_ping(xHttpCtx *ctx, void *arg) {
   (void)arg;
-  xHttpResponseSetHeader(writer, "Content-Type", "text/plain");
-  xHttpResponseSend(writer, "pong", 4);
+  xHttpCtxSetStatus(ctx, 200);
+  xHttpCtxSetHeader(ctx, "Content-Type", "text/plain");
+  xHttpCtxSend(ctx, "pong", 4);
 }
 
-// GET /echo?size=N → N bytes of 'x'
-// POST /echo → echo request body
-static void handle_echo(xHttpResponseWriter writer, const xHttpRequestConf *req, void *arg) {
+/* GET /echo?size=N → N bytes of 'x' */
+static void handle_echo(xHttpCtx *ctx, void *arg) {
   (void)arg;
 
-  if (strcmp(req->method, "POST") == 0) {
-    // Echo back the request body
-    xHttpResponseSetHeader(writer, "Content-Type", "application/octet-stream");
-    xHttpResponseSend(writer, req->body, req->body_len);
-    return;
-  }
-
-  // GET: parse ?size=N from URL
-  size_t      size = 64; // default
-  const char *q    = strchr(req->url, '?');
+  size_t      size = 64; /* default */
+  const char *q    = strchr(ctx->url, '?');
   if (q) {
     const char *sp = strstr(q, "size=");
     if (sp) {
       size = static_cast<size_t>(atoi(sp + 5));
       if (size == 0) size = 64;
-      if (size > 1048576) size = 1048576; // cap at 1MB
+      if (size > 1048576) size = 1048576; /* cap at 1MB */
     }
   }
 
   std::vector<char> body(size, 'x');
-  xHttpResponseSetHeader(writer, "Content-Type", "application/octet-stream");
-  xHttpResponseSend(writer, body.data(), body.size());
+  xHttpCtxSetStatus(ctx, 200);
+  xHttpCtxSetHeader(ctx, "Content-Type", "application/octet-stream");
+  xHttpCtxSend(ctx, body.data(), body.size());
+}
+
+/* POST /echo → echo the request body.
+ *
+ * Each request gets its own EchoBody via on_request (xHttpCtxSetUser);
+ * on_data accumulates chunks and on_done writes it. The EchoBody is
+ * freed by on_done on the happy path (and the user pointer nulled so
+ * on_close does not double-free); on_close frees it when the request
+ * dies before on_done (client disconnect / pending error / abort).
+ * Concurrent requests on the same route stay isolated because
+ * on_data/on_done see the per-request user data, not the route-level arg. */
+struct EchoBody {
+  std::vector<char> body;
+};
+
+static int echo_on_request(xHttpCtx *ctx, void *arg) {
+  (void)arg;
+  xHttpCtxSetUser(ctx, new EchoBody());
+  return 0;
+}
+
+static int echo_on_data(const char *data, size_t len, void *arg) {
+  auto *c = static_cast<EchoBody *>(arg);
+  c->body.insert(c->body.end(), data, data + len);
+  return 0;
+}
+
+static void echo_on_done(xHttpCtx *ctx, void *arg) {
+  auto *c = static_cast<EchoBody *>(arg);
+  xHttpCtxSetStatus(ctx, 200);
+  xHttpCtxSetHeader(ctx, "Content-Type", "application/octet-stream");
+  xHttpCtxSend(ctx, c->body.data(), c->body.size());
+  delete c;
+  xHttpCtxSetUser(ctx, NULL); /* on_close must not double-free */
+}
+
+/* Arg is the route-level arg (may be NULL) — read the per-request state
+ * from the ctx. on_close fires on EVERY teardown path (including ones that
+ * never reach on_done), so it is the leak guard for aborted requests. */
+static void echo_on_close(xHttpCtx *ctx, void *arg) {
+  (void)arg;
+  auto *c = static_cast<EchoBody *>(xHttpCtxUser(ctx));
+  if (c) {
+    delete c;
+    xHttpCtxSetUser(ctx, NULL);
+  }
 }
 
 int main(int argc, char *argv[]) {
@@ -75,24 +114,56 @@ int main(int argc, char *argv[]) {
     fprintf(stderr, "Failed to create event loop\n");
     return 1;
   }
+  xEventLoopEnter(g_loop); /* xHttpServerCreate requires a bound loop */
 
-  // Watch SIGINT to stop gracefully
-  xSignal(g_loop, SIGINT, [](int, void *) { xEventLoopStop(g_loop); }, nullptr);
+  /* Watch SIGINT to stop gracefully */
+  xSignal(SIGINT, [](int, void *) { xEventLoopStop(g_loop); }, nullptr);
 
-  xHttpServer server = xHttpServerCreate();
+  xHttpMux mux = xHttpMuxCreate();
+  if (!mux) {
+    fprintf(stderr, "Failed to create HTTP mux\n");
+    xEventLoopLeave();
+    xEventLoopDestroy(g_loop);
+    return 1;
+  }
+
+  xHttpServerConf conf = {};
+  conf.resolve         = xHttpMuxResolve;
+  conf.router          = mux;
+  conf.idle_timeout_ms = 60000;
+  xHttpServer server   = xHttpServerCreate(&conf);
   if (!server) {
     fprintf(stderr, "Failed to create HTTP server\n");
+    xHttpMuxDestroy(mux);
+    xEventLoopLeave();
     xEventLoopDestroy(g_loop);
     return 1;
   }
 
-  xHttpServerRoute(server, "GET /ping", handle_ping, nullptr);
-  xHttpServerRoute(server, "/echo", handle_echo, nullptr);
+  xHttpRouteConf rc = {};
+  rc.pattern        = "GET /ping";
+  rc.on_done        = handle_ping;
+  xHttpMuxHandle(mux, &rc);
+
+  rc         = {};
+  rc.pattern = "GET /echo";
+  rc.on_done = handle_echo;
+  xHttpMuxHandle(mux, &rc);
+
+  rc            = {};
+  rc.pattern    = "POST /echo";
+  rc.on_request = echo_on_request;
+  rc.on_data    = echo_on_data;
+  rc.on_done    = echo_on_done;
+  rc.on_close   = echo_on_close;
+  xHttpMuxHandle(mux, &rc);
 
   xErrno err = xHttpServerListen(server, "0.0.0.0", port);
   if (err != xErrno_Ok) {
     fprintf(stderr, "Failed to listen on port %u\n", port);
     xHttpServerDestroy(server);
+    xHttpMuxDestroy(mux);
+    xEventLoopLeave();
     xEventLoopDestroy(g_loop);
     return 1;
   }
@@ -107,6 +178,8 @@ int main(int argc, char *argv[]) {
   xEventLoopRun(g_loop, X_RUN_DEFAULT);
 
   xHttpServerDestroy(server);
+  xHttpMuxDestroy(mux);
+  xEventLoopLeave();
   xEventLoopDestroy(g_loop);
   return 0;
 }