From b8538e6ee2bff41819d7c044e6f9d287c5503222 Mon Sep 17 00:00:00 2001
From: Connor-Matthew <60215777+Connor-Matthew@users.noreply.github.com>
Date: Wed, 26 Aug 2026 17:58:06 +0800
Subject: [PATCH 1/4] [TransferEngine] Rebuild single Jetty on ACK timeout
 (status=9)

Avoid deleting the whole endpoint after a transient ACK timeout by draining
the faulty Jetty, recreating it locally, and rebinding to the existing peer
id. Falls back to deleteEndpoint on flush-done timeout or rebuild failure.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../jetty-ack-timeout-rebuild.md              |  15 +
 .../jetty-single-rebuild-plan.md              | 211 ++++++++++
 .../kunpeng_transport/urma/urma_endpoint.h    |  44 ++
 .../kunpeng_transport/urma/mock_urma.cpp      |  13 +-
 .../kunpeng_transport/urma/urma_endpoint.cpp  | 375 +++++++++++++++++-
 5 files changed, 647 insertions(+), 11 deletions(-)
 create mode 100644 docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md
 create mode 100644 docs/source/design/transfer-engine/jetty-single-rebuild-plan.md

diff --git a/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md b/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md
new file mode 100644
index 0000000000..4965d9c919
--- /dev/null
+++ b/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md
@@ -0,0 +1,15 @@
+---
+orphan: true
+---
+
+# Jetty ACK Timeout 重建方案（已废弃）
+
+**状态：已废弃。请勿按本文实现或评审。**
+
+权威方案见：
+
+[jetty-single-rebuild-plan.md](./jetty-single-rebuild-plan.md)
+
+废弃原因：本文把「对端 import/bind 同步」写成阻塞前置条件；经对照 kunpeng
+`UrmaEndpoint` 数据面（单边 READ/WRITE 打到 remote segment，不依赖对端 jetty
+收发语义）后，采用**本端单 Jetty 排空重建、不需要对端协议同步**的方案。
diff --git a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
new file mode 100644
index 0000000000..7fd20d67f8
--- /dev/null
+++ b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
@@ -0,0 +1,211 @@
+# Jetty ACK Timeout（status=9）单 Jetty 重建方案
+
+状态：实现中（分支 `feat/jetty-ack-timeout-single-rebuild`）— **权威方案**（取代已废弃的 `jetty-ack-timeout-rebuild.md`）
+范围：kunpeng / UB 传输路径（`UrmaEndpoint` / `UbWorkerPool`）；不涉及 tent
+关联错误码：
+
+| CR status | 枚举 | 本方案处理 |
+|---|---|---|
+| 4 | `URMA_CR_LOC_ACCESS_ERR` | 不重建；保持现有 slice 重试 / 失败逻辑 |
+| 9 | `URMA_CR_ACK_TIMEOUT_ERR` | 单 Jetty 排空 → 删除 → 重建，**纯本地操作** |
+
+---
+
+## 1. 背景与问题
+
+8 节点场景下，Jetty completion 出现 status=9（ACK 超时 / 重传超限）后，现有逻辑：
+
+1. poll 到非 SUCCESS → slice 进入重试
+2. 重试耗尽 → `deleteEndpointByPtr` → `UrmaEndpoint::deconstruct`
+3. `deconstruct` 直接 `unbind / unimport / delete_jetty`，**没有** ERROR 排空与 `FLUSH_ERR_DONE` 栅栏
+
+结果：短暂路径问题被放大成整 endpoint（甚至整节点）不可用，常需重启才能恢复。
+
+---
+
+## 2. 核心思路
+
+**只重建出问题的那个 jetty，不删整个 EP，且不需要对端同步协议。**
+
+```
+status=9 出现在 jetty_list_[i]
+  ① urma_modify_jetty(jetty_list_[i], ERROR)   ← 排空
+  ② poll 到 FLUSH_ERR_DONE(local_id)            ← 排空完成栅栏
+  ③ flush → unbind/unimport → delete → create
+     → import(原 peer id) → bind → 更新槽位     ← 安全替换（仍纯本地）
+```
+
+**为什么不需要对端同步？**
+
+kunpeng 数据面是单边 `URMA_OPC_READ/WRITE`，DMA 目标是对端 **segment**
+（`slice->ub.r_seg`），不是往对端 jetty 做 SEND/RECV。
+
+`urma_post_jetty_send_wr` 的第一个参数是**本地 jetty**。握手虽会
+`import` + `bind` 并对 `wr.tjetty` 赋值，但：
+
+- 现有路径在 `tjetty` 缺失时只 `LOG(ERROR)`，仍继续 post
+- `remote_jetty == NULL` 为空分支，未当硬失败
+- 重建后本端用**仍有效的对端 jetty id** 做 re-import/rebind 即可恢复本端发出方向；无需通知对端改其 import 视图
+
+反向流量若仍绑旧本端 id，对端可能自行报错并走自己的本地重建——两侧各自恢复，不上同步握手。
+
+---
+
+## 3. 状态机
+
+每个 jetty 槽位有独立状态：
+
+```
+ACTIVE ──(status=9)──► DRAINING ──(FLUSH_ERR_DONE)──► REBUILDING
+                                                          │
+                     ◄── create + 本端 rebind + 更新槽位 ──┘
+                     │
+                     └── 失败 / 超时 ──► 回退：deleteEndpointByPtr
+```
+
+- `ACTIVE`：正常收发
+- `DRAINING`：已 `modify(ERROR)`，禁止 post，等待 `FLUSH_ERR_DONE`
+- `REBUILDING`：排空完成，正在 flush / 删建 / rebind
+- 同 EP 任意时刻最多一个 jetty 处于 DRAINING/REBUILDING（串行）
+
+---
+
+## 4. 关键设计
+
+### 4.1 per-jetty 状态
+
+```cpp
+enum JettyState { ACTIVE, DRAINING, REBUILDING };
+std::vector<JettyState> jetty_state_;             // 与 jetty_list_ 平行
+std::unordered_map<uint32_t, int> jetty_id_map_;  // jetty_id → slot index
+std::vector<uint32_t> peer_jetty_id_;             // 每槽对端 id，delete 前保留
+```
+
+`peer_jetty_id_` 在握手 `doSetupConnection` 时写入；段 C 在 delete 本地 jetty
+前依赖它做 re-import，不得只依赖即将销毁的 `imported_jetty_map_` 指针键。
+
+### 4.2 选槽策略
+
+`submitPostSend` 选槽时跳过非 ACTIVE 槽：
+
+```
+随机选一个槽 → 非 ACTIVE 则重试 → 全部不可用则返回 0（上层重试）
+```
+
+### 4.3 锁策略
+
+| 阶段 | 是否持 `lock_` | 说明 |
+|---|---|---|
+| 段 A：modify(ERROR) | 是 | 与 post 互斥（UMDK 约束） |
+| 段 B：等 FLUSH_ERR_DONE | 否 | 由 performPoll 顺带消费，不占锁 |
+| 段 C：flush/删建/rebind | 是 | 替换 `jetty_list_` 与 `imported_jetty_map_` |
+
+### 4.4 FLUSH_ERR_DONE 消费
+
+走现有 `performPoll` → `UrmaContext::poll`。**必须在按 `user_ctx` 解 slice
+之前分流**（假 CQE 的 `user_ctx` 无效；现有空 ctx `continue` 会丢掉栅栏）：
+
+```
+if cr.status == FLUSH_ERR_DONE:
+    local_id → jetty_id_map_ → (endpoint, slot)
+    endpoint->onFlushDone(slot)  // DRAINING → REBUILDING，触发段 C
+    continue  // 假 CQE，不走 slice 路径，不改 outstanding slice 语义外的 depth 时需单独约定
+```
+
+### 4.5 status=9 分流与触发时机
+
+**首次** poll 到 `ACK_TIMEOUT_ERR (9)` 即触发排空（不等重试耗尽）；`onJettyError`
+必须幂等（已在 DRAINING/REBUILDING 则不再 `modify`）。
+
+```
+if cr.status == ACK_TIMEOUT_ERR (9):
+    slot ← slice->ub.jetty_depth 反查，或 cr.local_id → jetty_id_map_
+    slice->ub.endpoint → UrmaEndpoint
+    endpoint->onJettyError(slot)  // 段 A
+    slice 计入 retry（换 ACTIVE 槽重发）
+```
+
+### 4.6 outstanding 记账
+
+排空期间 inflight WR 会以 error/flush 类 CQE 回来。要求：
+
+- 带有效 `user_ctx` 的失败 CQE：仍走现有 `jetty_depth_set` / retry 路径扣
+  `wr_depth_list_` 与 JFC outstanding
+- `FLUSH_ERR_DONE`：不当作 slice；不得 `markSuccess` / 不得当失败 slice 入队
+- 段 C 替换前若 depth 仍非 0：打日志并在持锁下归零该槽与对应 JFC 计数（与今日
+  `deconstruct` 对 outstanding 的处理同思路），避免泄漏
+
+### 4.7 段 C 完整顺序（持锁）
+
+```
+1. urma_flush_jetty（回收残余 WR；注意与 poll 可能重复，见 §7）
+2. unbind → unimport（旧本地 jetty 上的对端视图）
+3. delete_jetty；从 jetty_id_map_ 删旧 id
+4. create_jetty（同 JFC/JFR 配置）
+5. import(peer_jetty_id_[slot]) → bind(新 jetty, imported)
+6. 更新 jetty_list_[slot]、imported_jetty_map_、jetty_id_map_
+7. jetty_state_[slot] = ACTIVE
+```
+
+任一步失败 → 回退 `deleteEndpointByPtr`。
+
+---
+
+## 5. 执行计划
+
+### Step 1：加状态与映射
+
+- `UrmaEndpoint` 增加 `jetty_state_`、`jetty_id_map_`、`peer_jetty_id_`
+- `construct()` / 握手初始化，`deconstruct()` 清理
+- `submitPostSend` 选槽跳过非 ACTIVE
+
+### Step 2：poll 分流
+
+- `UrmaContext::poll`：先识别 `FLUSH_ERR_DONE`（`local_id`），再处理
+  `ACK_TIMEOUT_ERR`（endpoint + 槽位）
+- 假 CQE 不走 slice 路径
+
+### Step 3：段 A — 触发排空
+
+- `onJettyError(slot)`：持锁 → 幂等检查 → `modify(ERROR)` → 标 DRAINING
+- 记录 drain 起始时间，供超时用
+
+### Step 4：段 B — 等待栅栏
+
+- `onFlushDone(slot)`：标 REBUILDING，触发段 C
+
+### Step 5：段 C — 重建
+
+- 按 §4.7 完整顺序执行
+
+### Step 6：超时降级
+
+- flush-done 等待超时（可配置，默认 3s）→ 回退 `deleteEndpointByPtr`
+- create / import / bind 失败同样降级
+
+### Step 7：日志与测试
+
+- 关键路径日志：`jetty_id`、slot、耗时、降级原因
+- 单测：状态机、选槽跳过、假 CQE 路由、幂等 `onJettyError`
+- 集成 / 故障注入：status=9 后该槽恢复 ACTIVE，同 EP 其它槽可继续；超时路径删 EP
+- 无 UMDK 硬件时，flush 与真实 ACK timeout 行为标为硬件验证项
+
+---
+
+## 6. 改动文件
+
+| 文件 | 改动 |
+|---|---|
+| `urma_endpoint.h` | `JettyState`、`jetty_state_`、`jetty_id_map_`、`peer_jetty_id_`、`onJettyError()` / `onFlushDone()` |
+| `urma_endpoint.cpp` | 状态机、选槽跳过、段 A/C、握手写入 `peer_jetty_id_` |
+| `ub_context.cpp` | poll / worker 侧配合 status=9 与 `FLUSH_ERR_DONE`（若分流落在 context poll 则改 `urma_endpoint.cpp` 中 `UrmaContext::poll`） |
+
+---
+
+## 7. 风险与开放问题
+
+1. **urma_flush_jetty 与 poll 重复**：flush 返回的 WR 级 CR 是否已在 JFC poll 中出现过，需实测确认，避免 double-complete。
+2. **共享 JFC 假 CQE 过滤**：多 jetty 共享 JFC 时，严格按 `local_id` 匹配，不能假设顺序。
+3. **超时阈值**：flush-done 等待 3s 是否合适，需结合 `err_timeout` 和现场标定。
+4. **硬件 hang 场景**：本方案解决软件放大故障；若根因是设备/驱动 hang，重建仍可能失败，保留删 EP 降级路径。
+5. **反向路径**：对端仍绑旧本端 id 时可能自行报 9 并本地重建；观察即可，本期不上对端协议。
diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
index 6b6fa7076a..d98b50993b 100644
--- a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
+++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
@@ -18,7 +18,10 @@
 #include <memory>
 #include <string>
 #include <thread>
+#include <unordered_map>
+#include <unordered_set>
 #include <utility>
+#include <vector>
 #include "common.h"
 #include "config.h"
 #include "urma_api.h"
@@ -47,6 +50,8 @@ static urma_import_seg_flag_t import_flag = {
            .reserved = 0}};
 
 // define the UrmaContext class
+class UrmaEndpoint;
+
 class UrmaContext : public UbContext {
     friend class UrmaEndpoint;
 
@@ -81,6 +86,15 @@ class UrmaContext : public UbContext {
     static bool uninit();
     static bool init();
 
+    void registerJettyOwner(uint32_t jetty_id, UrmaEndpoint* endpoint,
+                            int slot);
+    void unregisterJettyOwner(uint32_t jetty_id);
+    bool findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint,
+                        int* slot);
+    void addDrainingEndpoint(UrmaEndpoint* endpoint);
+    void removeDrainingEndpoint(UrmaEndpoint* endpoint);
+    void checkJettyDrainTimeouts();
+
    private:
     int construct(GlobalConfig& config) override;
     int deconstruct() override;
@@ -146,11 +160,21 @@ class UrmaContext : public UbContext {
 
     urma_import_seg_flag_t import_flag_ = mooncake::import_flag;
     std::unordered_map<std::string, urma_target_seg_t*> import_tseg_map;
+
+    RWSpinlock jetty_owner_lock_;
+    struct JettyOwner {
+        UrmaEndpoint* endpoint = nullptr;
+        int slot = -1;
+    };
+    std::unordered_map<uint32_t, JettyOwner> jetty_owner_map_;
+    std::unordered_set<UrmaEndpoint*> draining_endpoints_;
 };
 
 // define the UrmaEndpoint class
 class UrmaEndpoint : public UbEndPoint {
    public:
+    enum JettyState { ACTIVE = 0, DRAINING = 1, REBUILDING = 2 };
+
     UrmaEndpoint(UrmaContext* context)
         : context_(context), jfc_outstanding_(nullptr) {}
 
@@ -173,6 +197,13 @@ class UrmaEndpoint : public UbEndPoint {
 
     const std::string toString() const override;
 
+    // Called from UrmaContext::poll on ACK timeout / flush-done / drain timeout.
+    void onJettyError(int slot);
+    void onFlushDone(int slot);
+    void checkDrainTimeout();
+
+    int findSlotByDepth(volatile int* depth) const;
+
    private:
     void disconnectUnlocked() override;
 
@@ -187,6 +218,12 @@ class UrmaEndpoint : public UbEndPoint {
                           uint32_t peer_jetty_num,
                           std::string* reply_msg = nullptr);
 
+    bool hasNonActiveJettyUnlocked() const;
+    int selectActiveJettyUnlocked();
+    int rebuildJettyUnlocked(int slot);
+
+    static constexpr uint64_t kJettyDrainTimeoutNs = 3000000000ull;  // 3s
+
    private:
     UrmaContext* context_;
     urma_token_t urma_token = {.token = 0xACFE};
@@ -195,6 +232,13 @@ class UrmaEndpoint : public UbEndPoint {
     int max_wr_depth_;
     volatile int* jfc_outstanding_;
     std::unordered_map<urma_jetty_t*, urma_target_jetty_t*> imported_jetty_map_;
+
+    std::vector<JettyState> jetty_state_;
+    std::unordered_map<uint32_t, int> jetty_id_map_;
+    std::vector<uint32_t> peer_jetty_id_;
+    std::string peer_eid_;
+    uint64_t drain_start_ns_ = 0;
+    int draining_slot_ = -1;
 };
 }  // namespace mooncake
 #endif  // URMA_ENDPOINT_H
diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp
index c392577a6b..a05e04280c 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp
@@ -346,11 +346,12 @@ urma_jetty_t *urma_create_jetty(urma_context_t *ctx, urma_jetty_cfg_t *cfg) {
     if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) {
         return nullptr;
     }
+    static std::atomic<uint32_t> next_jetty_id{1};
     urma_jetty_t *jetty = new urma_jetty_t;
     memset(&jetty->jetty_id.eid, 0, sizeof(urma_eid_t));
     jetty->jetty_id.eid.raw[0] = 1;
     jetty->jetty_id.uasid = 0;
-    jetty->jetty_id.id = 1;
+    jetty->jetty_id.id = next_jetty_id.fetch_add(1);
     jetty->jetty_cfg = *cfg;
     jetty->remote_jetty = nullptr;
     jetty_map[jetty] = 1;
@@ -420,6 +421,16 @@ urma_status_t urma_modify_jetty(urma_jetty_t *jetty, urma_jetty_attr_t *attr) {
     return URMA_SUCCESS;
 }
 
+int urma_flush_jetty(urma_jetty_t *jetty, int cr_cnt, urma_cr_t *cr) {
+    (void)cr_cnt;
+    (void)cr;
+    std::shared_lock<std::shared_mutex> lock(g_rw_mutex);
+    if (!jetty || jetty_map.find(jetty) == jetty_map.end()) {
+        return -1;
+    }
+    return 0;
+}
+
 urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr,
                                       urma_jfs_wr_t **bad_wr) {
     {
diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
index 06c65d0385..7a9cda3cf6 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
@@ -518,11 +518,55 @@ bool UrmaContext::transEidFromString(const std::string& eid_str,
     return index == URMA_EID_SIZE;
 }
 
+void UrmaContext::registerJettyOwner(uint32_t jetty_id, UrmaEndpoint* endpoint,
+                                     int slot) {
+    RWSpinlock::WriteGuard guard(jetty_owner_lock_);
+    jetty_owner_map_[jetty_id] = JettyOwner{endpoint, slot};
+}
+
+void UrmaContext::unregisterJettyOwner(uint32_t jetty_id) {
+    RWSpinlock::WriteGuard guard(jetty_owner_lock_);
+    jetty_owner_map_.erase(jetty_id);
+}
+
+bool UrmaContext::findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint,
+                                 int* slot) {
+    RWSpinlock::ReadGuard guard(jetty_owner_lock_);
+    auto it = jetty_owner_map_.find(jetty_id);
+    if (it == jetty_owner_map_.end()) return false;
+    if (endpoint) *endpoint = it->second.endpoint;
+    if (slot) *slot = it->second.slot;
+    return true;
+}
+
+void UrmaContext::addDrainingEndpoint(UrmaEndpoint* endpoint) {
+    RWSpinlock::WriteGuard guard(jetty_owner_lock_);
+    draining_endpoints_.insert(endpoint);
+}
+
+void UrmaContext::removeDrainingEndpoint(UrmaEndpoint* endpoint) {
+    RWSpinlock::WriteGuard guard(jetty_owner_lock_);
+    draining_endpoints_.erase(endpoint);
+}
+
+void UrmaContext::checkJettyDrainTimeouts() {
+    std::vector<UrmaEndpoint*> endpoints;
+    {
+        RWSpinlock::ReadGuard guard(jetty_owner_lock_);
+        endpoints.assign(draining_endpoints_.begin(),
+                         draining_endpoints_.end());
+    }
+    for (auto* endpoint : endpoints) {
+        if (endpoint) endpoint->checkDrainTimeout();
+    }
+}
+
 int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
                       int& num_failed,
                       std::unordered_map<volatile int*, int>& jetty_depth_set,
                       int jfc_index) {
     num_failed = 0;
+    checkJettyDrainTimeouts();
     urma_cr_t cr[num_entries];
     int nr_poll = urma_poll_jfc(jfc_list_[jfc_index].native, num_entries, cr);
     if (nr_poll < 0) {
@@ -530,11 +574,26 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
                    << device_name_;
         return ERR_CONTEXT;
     }
+    int wr_completions = 0;
     for (int i = 0; i < nr_poll; ++i) {
+        // Fake fence CQE: user_ctx is invalid; match by local_id only.
+        if (cr[i].status == URMA_CR_WR_FLUSH_ERR_DONE) {
+            UrmaEndpoint* endpoint = nullptr;
+            int slot = -1;
+            if (findJettyOwner(cr[i].local_id, &endpoint, &slot) && endpoint) {
+                endpoint->onFlushDone(slot);
+            } else {
+                LOG(WARNING) << "FLUSH_ERR_DONE for unknown jetty local_id="
+                             << cr[i].local_id << " on " << device_name_;
+            }
+            continue;
+        }
+
         auto slice = (Transport::Slice*)cr[i].user_ctx;
         if (!slice) {
             continue;
         }
+        ++wr_completions;
 
         // All deref of `slice` (including the jetty_depth aggregation below)
         // MUST happen before markSuccess(): once that publishes completion,
@@ -553,6 +612,22 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
             continue;
         }
 
+        if (cr[i].status == URMA_CR_ACK_TIMEOUT_ERR) {
+            auto* endpoint = static_cast<UrmaEndpoint*>(slice->ub.endpoint);
+            if (endpoint) {
+                int slot = endpoint->findSlotByDepth(depth);
+                if (slot < 0) {
+                    UrmaEndpoint* mapped = nullptr;
+                    int mapped_slot = -1;
+                    if (findJettyOwner(cr[i].local_id, &mapped, &mapped_slot) &&
+                        mapped == endpoint) {
+                        slot = mapped_slot;
+                    }
+                }
+                if (slot >= 0) endpoint->onJettyError(slot);
+            }
+        }
+
         if (cr[i].status != URMA_CR_WR_FLUSH_ERR ||
             show_work_request_flushed_error_)
             LOG(ERROR) << "Worker: Process failed for slice (opcode: "
@@ -576,7 +651,8 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
         // safely deref it.
         failed_slices[num_failed++] = slice;
     }
-    return nr_poll;
+    // Exclude FLUSH_ERR_DONE from outstanding accounting (it was never posted).
+    return wr_completions;
 }
 
 volatile int* UrmaContext::outstandingCount(int jfc_index) {
@@ -626,6 +702,12 @@ int UrmaEndpoint::construct(GlobalConfig& config) {
     }
 
     jetty_list_.resize(num_jetty_list);
+    jetty_state_.assign(num_jetty_list, ACTIVE);
+    peer_jetty_id_.assign(num_jetty_list, 0);
+    jetty_id_map_.clear();
+    peer_eid_.clear();
+    drain_start_ns_ = 0;
+    draining_slot_ = -1;
     auto* jfc = context_->jfc();
     jfc_outstanding_ = (volatile int*)jfc->jfc_cfg.user_ctx;
 
@@ -660,8 +742,11 @@ int UrmaEndpoint::construct(GlobalConfig& config) {
             PLOG(ERROR) << "Failed to create jetty";
             return ERR_ENDPOINT;
         }
-        LOG(INFO) << "Create jetty success, jetty id = "
-                  << jetty_list_[i]->jetty_id.id << " ,jetty jfc id = "
+        uint32_t jetty_id = jetty_list_[i]->jetty_id.id;
+        jetty_id_map_[jetty_id] = static_cast<int>(i);
+        context_->registerJettyOwner(jetty_id, this, static_cast<int>(i));
+        LOG(INFO) << "Create jetty success, jetty id = " << jetty_id
+                  << " ,jetty jfc id = "
                   << jetty_list_[i]->jetty_cfg.jfs_cfg.jfc->jfc_id.id << " : "
                   << jfc->jfc_id.id;
     }
@@ -672,19 +757,25 @@ int UrmaEndpoint::construct(GlobalConfig& config) {
 
 int UrmaEndpoint::deconstruct() {
     int ret = 0;
+    context_->removeDrainingEndpoint(this);
     for (size_t i = 0; i < jetty_list_.size(); ++i) {
         auto imported_it = imported_jetty_map_.find(jetty_list_[i]);
         auto imported_jetty = (imported_it != imported_jetty_map_.end())
                                   ? imported_it->second
                                   : nullptr;
-        ret = urma_unbind_jetty(jetty_list_[i]);
-        if (ret) PLOG(ERROR) << "Failed to unbind jetty";
+        if (jetty_list_[i]) {
+            context_->unregisterJettyOwner(jetty_list_[i]->jetty_id.id);
+            ret = urma_unbind_jetty(jetty_list_[i]);
+            if (ret) PLOG(ERROR) << "Failed to unbind jetty";
+        }
         if (imported_jetty != nullptr) {
             ret = urma_unimport_jetty(imported_jetty);
             if (ret) PLOG(ERROR) << "Failed to unimport jetty";
         }
-        ret = urma_delete_jetty(jetty_list_[i]);
-        if (ret) PLOG(ERROR) << "Failed to delete jetty";
+        if (jetty_list_[i]) {
+            ret = urma_delete_jetty(jetty_list_[i]);
+            if (ret) PLOG(ERROR) << "Failed to delete jetty";
+        }
         // After destroying QP, the wr_depth_list_ won't change
         bool displayed = false;
         if (wr_depth_list_[i] != 0) {
@@ -698,7 +789,14 @@ int UrmaEndpoint::deconstruct() {
         }
     }
     jetty_list_.clear();
+    jetty_state_.clear();
+    jetty_id_map_.clear();
+    peer_jetty_id_.clear();
+    peer_eid_.clear();
+    draining_slot_ = -1;
+    drain_start_ns_ = 0;
     delete[] wr_depth_list_;
+    wr_depth_list_ = nullptr;
     imported_jetty_map_.clear();
     return 0;
 }
@@ -791,8 +889,12 @@ int UrmaEndpoint::setupConnectionsByActive() {
 void UrmaEndpoint::disconnectUnlocked() {
     urma_jetty_attr_t attr;
     memset(&attr, 0, sizeof(attr));
+    attr.mask = JETTY_STATE;
     attr.state = URMA_JETTY_STATE_RESET;
 
+    context_->removeDrainingEndpoint(this);
+    draining_slot_ = -1;
+    drain_start_ns_ = 0;
     for (size_t i = 0; i < jetty_list_.size(); ++i) {
         int ret = urma_modify_jetty(jetty_list_[i], &attr);
         if (ret) PLOG(ERROR) << "Failed to modify jetty to RESET";
@@ -812,7 +914,11 @@ void UrmaEndpoint::disconnectUnlocked() {
             __sync_fetch_and_sub(jfc_outstanding_, wr_depth_list_[i]);
             wr_depth_list_[i] = 0;
         }
+        jetty_state_[i] = ACTIVE;
     }
+    imported_jetty_map_.clear();
+    peer_jetty_id_.assign(jetty_list_.size(), 0);
+    peer_eid_.clear();
     status_.store(UNCONNECTED, std::memory_order_release);
 }
 
@@ -873,7 +979,8 @@ int UrmaEndpoint::submitPostSend(
     std::vector<Transport::Slice*>& failed_slice_list) {
     RWSpinlock::WriteGuard guard(lock_);
     if (!active_) return 0;
-    int jetty_index = SimpleRandom::Get().next(jetty_list_.size());
+    int jetty_index = selectActiveJettyUnlocked();
+    if (jetty_index < 0) return 0;
     int wr_count = std::min(max_wr_depth_ - wr_depth_list_[jetty_index],
                             (int)slice_list.size());
     wr_count =
@@ -929,8 +1036,6 @@ int UrmaEndpoint::submitPostSend(
     }
     __sync_fetch_and_add(&wr_depth_list_[jetty_index], wr_count);
     __sync_fetch_and_add(jfc_outstanding_, wr_count);
-    if (jetty_list_[jetty_index]->remote_jetty == NULL) {
-    }
     int rc =
         urma_post_jetty_send_wr(jetty_list_[jetty_index], wr_list, &bad_wr);
     if (rc) {
@@ -968,6 +1073,7 @@ int UrmaEndpoint::doSetupConnection(const std::string& peer_eid,
         return ERR_INVALID_ARGUMENT;
     }
 
+    peer_eid_ = peer_eid;
     for (int jetty_index = 0; jetty_index < (int)jetty_list_.size();
          ++jetty_index) {
         int ret = doSetupConnection(
@@ -1011,12 +1117,261 @@ int UrmaEndpoint::doSetupConnection(int jetty_index,
         return ERR_ENDPOINT;
     }
     imported_jetty_map_[jetty] = imported_jetty;
+    peer_jetty_id_[jetty_index] = peer_jetty_num;
+    jetty_state_[jetty_index] = ACTIVE;
     LOG(INFO) << "Bind jetty success, local jetty id:" << jetty->jetty_id.id
               << ", remote jetty id:" << peer_jetty_num;
 
     return 0;
 }
 
+int UrmaEndpoint::findSlotByDepth(volatile int* depth) const {
+    if (!depth || !wr_depth_list_) return -1;
+    for (size_t i = 0; i < jetty_list_.size(); ++i) {
+        if (&wr_depth_list_[i] == depth) return static_cast<int>(i);
+    }
+    return -1;
+}
+
+bool UrmaEndpoint::hasNonActiveJettyUnlocked() const {
+    for (auto state : jetty_state_) {
+        if (state != ACTIVE) return true;
+    }
+    return false;
+}
+
+int UrmaEndpoint::selectActiveJettyUnlocked() {
+    if (jetty_list_.empty()) return -1;
+    const int n = static_cast<int>(jetty_list_.size());
+    int start = SimpleRandom::Get().next(n);
+    for (int i = 0; i < n; ++i) {
+        int idx = (start + i) % n;
+        if (jetty_state_[idx] == ACTIVE) return idx;
+    }
+    return -1;
+}
+
+void UrmaEndpoint::onJettyError(int slot) {
+    bool delete_ep = false;
+    {
+        RWSpinlock::WriteGuard guard(lock_);
+        if (slot < 0 || slot >= static_cast<int>(jetty_list_.size())) return;
+        if (jetty_state_[slot] == DRAINING || jetty_state_[slot] == REBUILDING) {
+            return;  // idempotent
+        }
+        // Serial rebuild: at most one non-ACTIVE jetty per endpoint.
+        if (hasNonActiveJettyUnlocked()) {
+            LOG(INFO) << "Skip jetty rebuild for slot " << slot
+                      << ": another jetty is already draining/rebuilding on "
+                      << toString();
+            return;
+        }
+        if (!jetty_list_[slot]) return;
+
+        urma_jetty_attr_t attr{};
+        attr.mask = JETTY_STATE;
+        attr.state = URMA_JETTY_STATE_ERROR;
+        int ret = urma_modify_jetty(jetty_list_[slot], &attr);
+        if (ret) {
+            PLOG(ERROR) << "Failed to modify jetty to ERROR, slot=" << slot
+                        << " jetty_id=" << jetty_list_[slot]->jetty_id.id;
+            context_->removeDrainingEndpoint(this);
+            draining_slot_ = -1;
+            drain_start_ns_ = 0;
+            delete_ep = true;
+        } else {
+            jetty_state_[slot] = DRAINING;
+            draining_slot_ = slot;
+            drain_start_ns_ = getCurrentTimeInNano();
+            context_->addDrainingEndpoint(this);
+            LOG(WARNING) << "Jetty ACK timeout: start drain slot=" << slot
+                         << " jetty_id=" << jetty_list_[slot]->jetty_id.id
+                         << " on " << toString();
+        }
+    }
+    if (delete_ep) {
+        LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
+                   << "modify_jetty(ERROR) failed on " << toString();
+        context_->deleteEndpointByPtr(this);
+    }
+}
+
+void UrmaEndpoint::onFlushDone(int slot) {
+    bool delete_ep = false;
+    {
+        RWSpinlock::WriteGuard guard(lock_);
+        if (slot < 0 || slot >= static_cast<int>(jetty_list_.size())) return;
+        if (jetty_state_[slot] != DRAINING) return;
+        jetty_state_[slot] = REBUILDING;
+        LOG(INFO) << "Jetty flush-done: rebuild slot=" << slot << " on "
+                  << toString();
+        if (rebuildJettyUnlocked(slot)) {
+            context_->removeDrainingEndpoint(this);
+            draining_slot_ = -1;
+            drain_start_ns_ = 0;
+            delete_ep = true;
+        }
+    }
+    if (delete_ep) {
+        LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
+                   << "rebuildJetty failed on " << toString();
+        context_->deleteEndpointByPtr(this);
+    }
+}
+
+void UrmaEndpoint::checkDrainTimeout() {
+    bool delete_ep = false;
+    {
+        RWSpinlock::WriteGuard guard(lock_);
+        if (draining_slot_ < 0) return;
+        int slot = draining_slot_;
+        if (slot >= static_cast<int>(jetty_state_.size()) ||
+            jetty_state_[slot] != DRAINING) {
+            return;
+        }
+        uint64_t now = getCurrentTimeInNano();
+        if (now - drain_start_ns_ < kJettyDrainTimeoutNs) return;
+        LOG(ERROR) << "Jetty drain timed out after "
+                   << ((now - drain_start_ns_) / 1000000ull)
+                   << "ms, slot=" << slot << " on " << toString();
+        context_->removeDrainingEndpoint(this);
+        draining_slot_ = -1;
+        drain_start_ns_ = 0;
+        delete_ep = true;
+    }
+    if (delete_ep) {
+        LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
+                   << "flush-done timeout on " << toString();
+        context_->deleteEndpointByPtr(this);
+    }
+}
+
+int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
+    auto* old_jetty = jetty_list_[slot];
+    if (!old_jetty) return ERR_ENDPOINT;
+    const uint32_t old_id = old_jetty->jetty_id.id;
+    const uint32_t peer_id = peer_jetty_id_[slot];
+    const uint64_t started_ns = drain_start_ns_;
+    urma_jfc_t* reuse_jfc = old_jetty->jetty_cfg.jfs_cfg.jfc;
+    urma_jfr_t* reuse_jfr = old_jetty->jetty_cfg.shared.jfr;
+
+    // 1) Flush residual WRs (may overlap with already-polled CRs).
+    urma_cr_t flush_crs[64];
+    while (true) {
+        int flushed = urma_flush_jetty(old_jetty, 64, flush_crs);
+        if (flushed < 0) {
+            PLOG(ERROR) << "urma_flush_jetty failed, slot=" << slot;
+            return ERR_ENDPOINT;
+        }
+        if (flushed == 0) break;
+        // Completions for these WRs should already have been (or will be)
+        // accounted via poll; do not touch slice pointers from flush CRs.
+    }
+
+    // 2) Unbind / unimport old peer view.
+    auto imported_it = imported_jetty_map_.find(old_jetty);
+    urma_target_jetty_t* old_imported =
+        (imported_it != imported_jetty_map_.end()) ? imported_it->second
+                                                   : nullptr;
+    int ret = urma_unbind_jetty(old_jetty);
+    if (ret) PLOG(ERROR) << "Failed to unbind jetty before rebuild";
+    if (old_imported) {
+        ret = urma_unimport_jetty(old_imported);
+        if (ret) PLOG(ERROR) << "Failed to unimport jetty before rebuild";
+        imported_jetty_map_.erase(imported_it);
+    }
+
+    // 3) Delete old jetty and clear outstanding depth for this slot.
+    context_->unregisterJettyOwner(old_id);
+    jetty_id_map_.erase(old_id);
+    ret = urma_delete_jetty(old_jetty);
+    if (ret) {
+        PLOG(ERROR) << "Failed to delete jetty during rebuild";
+        jetty_list_[slot] = nullptr;
+        return ERR_ENDPOINT;
+    }
+    jetty_list_[slot] = nullptr;
+    if (wr_depth_list_[slot] != 0) {
+        __sync_fetch_and_sub(jfc_outstanding_, wr_depth_list_[slot]);
+        wr_depth_list_[slot] = 0;
+    }
+
+    // 4) Create replacement jetty with the same JFC/JFR config.
+    urma_jfs_cfg_t jfs_cfg = {
+        .depth = 2048,
+        .trans_mode = URMA_TM_RC,
+        .priority = 15,
+        .max_sge = 5,
+        .rnr_retry = 7,
+        .err_timeout = 17,
+        .user_ctx = 0,
+    };
+    urma_jetty_flag_t jetty_flag = {};
+    jetty_flag.bs.share_jfr = 1;
+    urma_jetty_cfg_t attr{};
+    attr.flag = jetty_flag;
+    attr.jfs_cfg = jfs_cfg;
+    attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc();
+    attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr();
+    urma_jetty_t* new_jetty =
+        urma_create_jetty(context_->urma_context_, &attr);
+    if (!new_jetty) {
+        PLOG(ERROR) << "Failed to create jetty during rebuild";
+        return ERR_ENDPOINT;
+    }
+
+    // 5) Re-import peer and bind locally (no peer protocol).
+    if (peer_eid_.empty()) {
+        LOG(ERROR) << "Missing peer eid during jetty rebuild";
+        urma_delete_jetty(new_jetty);
+        return ERR_ENDPOINT;
+    }
+    urma_eid_t eid;
+    if (!context_->transEidFromString(peer_eid_, eid)) {
+        LOG(ERROR) << "Invalid peer eid during jetty rebuild: " << peer_eid_;
+        urma_delete_jetty(new_jetty);
+        return ERR_ENDPOINT;
+    }
+    urma_rjetty_t rjetty = {};
+    rjetty.jetty_id.id = peer_id;
+    rjetty.jetty_id.eid = eid;
+    rjetty.trans_mode = URMA_TM_RC;
+    rjetty.type = URMA_JETTY;
+    rjetty.tp_type = URMA_CTP;
+    rjetty.flag.value = 0;
+    urma_target_jetty_t* imported =
+        urma_import_jetty(context_->urma_context_, &rjetty, &urma_token);
+    if (!imported) {
+        PLOG(ERROR) << "Failed to import peer jetty during rebuild";
+        urma_delete_jetty(new_jetty);
+        return ERR_ENDPOINT;
+    }
+    urma_status_t bind_ret = urma_bind_jetty(new_jetty, imported);
+    if (bind_ret != URMA_SUCCESS && bind_ret != URMA_EEXIST) {
+        PLOG(ERROR) << "Failed to bind rebuilt jetty";
+        urma_unimport_jetty(imported);
+        urma_delete_jetty(new_jetty);
+        return ERR_ENDPOINT;
+    }
+
+    jetty_list_[slot] = new_jetty;
+    imported_jetty_map_[new_jetty] = imported;
+    const uint32_t new_id = new_jetty->jetty_id.id;
+    jetty_id_map_[new_id] = slot;
+    context_->registerJettyOwner(new_id, this, slot);
+    jetty_state_[slot] = ACTIVE;
+    draining_slot_ = -1;
+    drain_start_ns_ = 0;
+    context_->removeDrainingEndpoint(this);
+
+    LOG(WARNING) << "Jetty rebuilt successfully slot=" << slot
+                 << " old_id=" << old_id << " new_id=" << new_id
+                 << " peer_id=" << peer_id << " elapsed_ms="
+                 << ((getCurrentTimeInNano() - started_ns) / 1000000ull)
+                 << " on " << toString();
+    return 0;
+}
+
 std::shared_ptr<UbEndPoint> UrmaContext::makeEndpoint() {
     return std::make_shared<UrmaEndpoint>(this);
 }

From 115191f24b8c1f307209daff93b5b3a55294ca4e Mon Sep 17 00:00:00 2001
From: Connor-Matthew <60215777+Connor-Matthew@users.noreply.github.com>
Date: Wed, 26 Aug 2026 19:20:53 +0800
Subject: [PATCH 2/4] [TransferEngine] Fix CI format and docs orphan metadata

Mark the jetty rebuild design doc as Sphinx orphan and apply clang-format
to the UrmaEndpoint changes so PR checks pass.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../design/transfer-engine/jetty-single-rebuild-plan.md     | 4 ++++
 .../transport/kunpeng_transport/urma/urma_endpoint.h        | 6 +++---
 .../src/transport/kunpeng_transport/urma/urma_endpoint.cpp  | 6 +++---
 3 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
index 7fd20d67f8..85ec8b0e0b 100644
--- a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
+++ b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
@@ -1,3 +1,7 @@
+---
+orphan: true
+---
+
 # Jetty ACK Timeout（status=9）单 Jetty 重建方案
 
 状态：实现中（分支 `feat/jetty-ack-timeout-single-rebuild`）— **权威方案**（取代已废弃的 `jetty-ack-timeout-rebuild.md`）
diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
index d98b50993b..af456e88e0 100644
--- a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
+++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
@@ -89,8 +89,7 @@ class UrmaContext : public UbContext {
     void registerJettyOwner(uint32_t jetty_id, UrmaEndpoint* endpoint,
                             int slot);
     void unregisterJettyOwner(uint32_t jetty_id);
-    bool findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint,
-                        int* slot);
+    bool findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint, int* slot);
     void addDrainingEndpoint(UrmaEndpoint* endpoint);
     void removeDrainingEndpoint(UrmaEndpoint* endpoint);
     void checkJettyDrainTimeouts();
@@ -197,7 +196,8 @@ class UrmaEndpoint : public UbEndPoint {
 
     const std::string toString() const override;
 
-    // Called from UrmaContext::poll on ACK timeout / flush-done / drain timeout.
+    // Called from UrmaContext::poll on ACK timeout / flush-done / drain
+    // timeout.
     void onJettyError(int slot);
     void onFlushDone(int slot);
     void checkDrainTimeout();
diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
index 7a9cda3cf6..14a317f3d9 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
@@ -1156,7 +1156,8 @@ void UrmaEndpoint::onJettyError(int slot) {
     {
         RWSpinlock::WriteGuard guard(lock_);
         if (slot < 0 || slot >= static_cast<int>(jetty_list_.size())) return;
-        if (jetty_state_[slot] == DRAINING || jetty_state_[slot] == REBUILDING) {
+        if (jetty_state_[slot] == DRAINING ||
+            jetty_state_[slot] == REBUILDING) {
             return;  // idempotent
         }
         // Serial rebuild: at most one non-ACTIVE jetty per endpoint.
@@ -1313,8 +1314,7 @@ int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
     attr.jfs_cfg = jfs_cfg;
     attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc();
     attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr();
-    urma_jetty_t* new_jetty =
-        urma_create_jetty(context_->urma_context_, &attr);
+    urma_jetty_t* new_jetty = urma_create_jetty(context_->urma_context_, &attr);
     if (!new_jetty) {
         PLOG(ERROR) << "Failed to create jetty during rebuild";
         return ERR_ENDPOINT;

From 60e947da6cdc52dfd09192e5d587ae295c750214 Mon Sep 17 00:00:00 2001
From: Connor-Matthew <60215777+Connor-Matthew@users.noreply.github.com>
Date: Thu, 27 Aug 2026 00:41:18 +0800
Subject: [PATCH 3/4] [TransferEngine] Fix jetty rebuild safety issues on ACK
 timeout path

Defer endpoint deletion until after poll depth accounting to avoid UAF,
deliver flush completions during rebuild, and isolate stale CQEs with
per-slot jetty epochs. Align disconnect teardown with ERROR flush fence
constraints and document the updated rebuild accounting model.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../jetty-single-rebuild-plan.md              |  48 ++-
 .../transport/kunpeng_transport/ub_context.h  |  22 +-
 .../kunpeng_transport/urma/urma_endpoint.h    |  62 ++-
 .../include/transport/transport.h             |   1 +
 .../kunpeng_transport/ub_context.cpp          |  28 +-
 .../kunpeng_transport/urma/urma_endpoint.cpp  | 395 ++++++++++++------
 6 files changed, 382 insertions(+), 174 deletions(-)

diff --git a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
index 85ec8b0e0b..956cb201ce 100644
--- a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
+++ b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md
@@ -60,16 +60,21 @@ kunpeng 数据面是单边 `URMA_OPC_READ/WRITE`，DMA 目标是对端 **segment
 每个 jetty 槽位有独立状态：
 
 ```
-ACTIVE ──(status=9)──► DRAINING ──(FLUSH_ERR_DONE)──► REBUILDING
-                                                          │
-                     ◄── create + 本端 rebind + 更新槽位 ──┘
-                     │
-                     └── 失败 / 超时 ──► 回退：deleteEndpointByPtr
+ACTIVE ──(status=9，无其它槽在重建)──► DRAINING ──(FLUSH_ERR_DONE)──► REBUILDING
+  │                                                                    │
+  │                 ◄── create + 本端 rebind + 更新槽位 ───────────────┘
+  │
+  └──(status=9，已有槽在重建)──► PENDING_DRAIN ──(前槽重建完成后触发)──► DRAINING
+
+                  任一环节失败 / 超时 ──► 回退：deleteEndpointByPtr
 ```
 
 - `ACTIVE`：正常收发
 - `DRAINING`：已 `modify(ERROR)`，禁止 post，等待 `FLUSH_ERR_DONE`
 - `REBUILDING`：排空完成，正在 flush / 删建 / rebind
+- `PENDING_DRAIN`：本槽也收到 status=9，但同 EP 已有 jetty 在重建；禁止 post
+  （选槽跳过，避免故障槽继续吃流量），等前一个重建完成后由其尾部串行触发
+  `modify(ERROR)` 进入 DRAINING
 - 同 EP 任意时刻最多一个 jetty 处于 DRAINING/REBUILDING（串行）
 
 ---
@@ -79,10 +84,11 @@ ACTIVE ──(status=9)──► DRAINING ──(FLUSH_ERR_DONE)──► REBUIL
 ### 4.1 per-jetty 状态
 
 ```cpp
-enum JettyState { ACTIVE, DRAINING, REBUILDING };
+enum JettyState { ACTIVE, DRAINING, REBUILDING, PENDING_DRAIN };
 std::vector<JettyState> jetty_state_;             // 与 jetty_list_ 平行
 std::unordered_map<uint32_t, int> jetty_id_map_;  // jetty_id → slot index
 std::vector<uint32_t> peer_jetty_id_;             // 每槽对端 id，delete 前保留
+std::vector<uint64_t> jetty_epoch_;               // 每槽重建代次，丢弃旧代次的迟到 CR
 ```
 
 `peer_jetty_id_` 在握手 `doSetupConnection` 时写入；段 C 在 delete 本地 jetty
@@ -119,7 +125,9 @@ if cr.status == FLUSH_ERR_DONE:
 ### 4.5 status=9 分流与触发时机
 
 **首次** poll 到 `ACK_TIMEOUT_ERR (9)` 即触发排空（不等重试耗尽）；`onJettyError`
-必须幂等（已在 DRAINING/REBUILDING 则不再 `modify`）。
+必须幂等（已在 DRAINING/REBUILDING/PENDING_DRAIN 则不再 `modify`）。
+若同 EP 已有 jetty 在重建，则把本槽标记为 `PENDING_DRAIN` 排队：选槽立即跳过
+它，等前一个重建完成后由其尾部调用 `startDrainUnlocked` 串行启动排空。
 
 ```
 if cr.status == ACK_TIMEOUT_ERR (9):
@@ -136,19 +144,29 @@ if cr.status == ACK_TIMEOUT_ERR (9):
 - 带有效 `user_ctx` 的失败 CQE：仍走现有 `jetty_depth_set` / retry 路径扣
   `wr_depth_list_` 与 JFC outstanding
 - `FLUSH_ERR_DONE`：不当作 slice；不得 `markSuccess` / 不得当失败 slice 入队
-- 段 C 替换前若 depth 仍非 0：打日志并在持锁下归零该槽与对应 JFC 计数（与今日
-  `deconstruct` 对 outstanding 的处理同思路），避免泄漏
+- **每个 WR 恰好完成一次**：`modify(ERROR)` 后 inflight WR 要么以
+  `WR_FLUSH_ERR` 经 JFC poll 回来，要么被段 C 的 `urma_flush_jetty` 回收；两者
+  统一经 `processWrCompletion` 记账（`jetty_depth_set` 延迟扣减 + poll 返回值
+  累计 JFC outstanding）。因此段 C 删除旧 jetty 后**不做**额外的
+  `wr_depth_list_[slot]` / JFC 清零——延迟扣减要等 poll 返回后才 apply，在段 C
+  里提前清零会双扣（实现后已删除原方案中的"归零兜底"）
+- 旧 jetty 删除后 `++jetty_epoch_[slot]`；slice 在 post 时记录
+  `slice->ub.jetty_epoch`，旧代次的迟到/重复 CR 在 `processWrCompletion` 里按
+  epoch 不匹配直接丢弃，不参与记账、不再入 retry 队列
 
 ### 4.7 段 C 完整顺序（持锁）
 
 ```
-1. urma_flush_jetty（回收残余 WR；注意与 poll 可能重复，见 §7）
+1. urma_flush_jetty（回收残余 WR，逐条经 processWrCompletion 交付；
+   注意与 poll 可能重复，见 §7）
 2. unbind → unimport（旧本地 jetty 上的对端视图）
-3. delete_jetty；从 jetty_id_map_ 删旧 id
+3. delete_jetty；从 jetty_id_map_ 删旧 id；++jetty_epoch_[slot]
 4. create_jetty（同 JFC/JFR 配置）
 5. import(peer_jetty_id_[slot]) → bind(新 jetty, imported)
 6. 更新 jetty_list_[slot]、imported_jetty_map_、jetty_id_map_
 7. jetty_state_[slot] = ACTIVE
+8. 扫描 PENDING_DRAIN 槽：有则立即 startDrainUnlocked（modify(ERROR)），
+   仍保持同 EP 串行
 ```
 
 任一步失败 → 回退 `deleteEndpointByPtr`。
@@ -200,7 +218,7 @@ if cr.status == ACK_TIMEOUT_ERR (9):
 
 | 文件 | 改动 |
 |---|---|
-| `urma_endpoint.h` | `JettyState`、`jetty_state_`、`jetty_id_map_`、`peer_jetty_id_`、`onJettyError()` / `onFlushDone()` |
+| `urma_endpoint.h` | `JettyState`（含 `PENDING_DRAIN`）、`jetty_state_`、`jetty_id_map_`、`peer_jetty_id_`、`jetty_epoch_`、`onJettyError()` / `onFlushDone()` / `startDrainUnlocked()` |
 | `urma_endpoint.cpp` | 状态机、选槽跳过、段 A/C、握手写入 `peer_jetty_id_` |
 | `ub_context.cpp` | poll / worker 侧配合 status=9 与 `FLUSH_ERR_DONE`（若分流落在 context poll 则改 `urma_endpoint.cpp` 中 `UrmaContext::poll`） |
 
@@ -208,7 +226,11 @@ if cr.status == ACK_TIMEOUT_ERR (9):
 
 ## 7. 风险与开放问题
 
-1. **urma_flush_jetty 与 poll 重复**：flush 返回的 WR 级 CR 是否已在 JFC poll 中出现过，需实测确认，避免 double-complete。
+1. **urma_flush_jetty 与 poll 重复**：flush 返回的 WR 级 CR 是否已在 JFC poll
+   中出现过，需实测确认，避免 double-complete。实现按「每个 WR 恰好完成一次」
+   记账，并用 `jetty_epoch_` 把旧代次的迟到 CR 整体丢弃兜底；若实测发现 flush
+   与 poll 会重复交付同一 WR，需重新评估 slice 指针解引用的安全性（届时 CR 里
+   的 `user_ctx` 可能指向已回收的 slice）。
 2. **共享 JFC 假 CQE 过滤**：多 jetty 共享 JFC 时，严格按 `local_id` 匹配，不能假设顺序。
 3. **超时阈值**：flush-done 等待 3s 是否合适，需结合 `err_timeout` 和现场标定。
 4. **硬件 hang 场景**：本方案解决软件放大故障；若根因是设备/驱动 hang，重建仍可能失败，保留删 EP 降级路径。
diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h
index 56767c9dee..db61ca1985 100644
--- a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h
+++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h
@@ -188,13 +188,19 @@ class UbContext {
     //   * Successful slices have markSuccess() called in place and are NOT
     //     returned (they may be recycled by the submitting thread the
     //     moment markSuccess() runs).
-    //   * Failed slices are returned in failed_slices[0..num_failed-1] for
-    //     the caller to apply retry / markFailed.
-    // Returns the total number of completions polled (>= 0), or a negative
-    // error code.
-    virtual int poll(int num_entries, Transport::Slice** failed_slices,
-                     int& num_failed,
+    //   * Failed slices are appended to failed_slices for the caller to apply
+    //     retry / markFailed. Implementations may also append completions
+    //     recovered while draining a jetty, so the vector is unbounded.
+    //   * Stale completions from a previous jetty generation are dropped and
+    //     are not counted in the return value.
+    //   * Endpoints scheduled for deletion are appended to deferred_deletes;
+    //     the caller must delete them only after jetty_depth_set accounting.
+    // Returns the number of resolved WR completions (>= 0), or a negative
+    // error code. Fence markers and dropped stale completions are excluded.
+    virtual int poll(int num_entries,
+                     std::vector<Transport::Slice*>& failed_slices,
                      std::unordered_map<volatile int*, int>& jetty_depth_set,
+                     std::vector<UbEndPoint*>& deferred_deletes,
                      int jfc_index = 0) = 0;
 
     virtual volatile int* outstandingCount(int jfc_index) = 0;
@@ -259,6 +265,10 @@ class UbContext {
 
     UbTransport& engine() const { return engine_; }
 
+    bool traceWorkRequestFlushedErrors() const {
+        return show_work_request_flushed_error_;
+    }
+
     uint8_t portNum() const { return port_; }
 
     int activeSpeed() const { return active_speed_; }
diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
index af456e88e0..25d3e71a2d 100644
--- a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
+++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h
@@ -63,8 +63,9 @@ class UrmaContext : public UbContext {
     int unregisterMemoryRegion(uint64_t va) override;
     int doProcessContextEvents() override;
     void* retrieveRemoteSeg(const std::string& value) override;
-    int poll(int num_entries, Transport::Slice** failed_slices, int& num_failed,
+    int poll(int num_entries, std::vector<Transport::Slice*>& failed_slices,
              std::unordered_map<volatile int*, int>& jetty_depth_set,
+             std::vector<UbEndPoint*>& deferred_deletes,
              int jfc_index) override;
     volatile int* outstandingCount(int jfc_index) override;
     int submitPostSend(
@@ -92,7 +93,10 @@ class UrmaContext : public UbContext {
     bool findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint, int* slot);
     void addDrainingEndpoint(UrmaEndpoint* endpoint);
     void removeDrainingEndpoint(UrmaEndpoint* endpoint);
-    void checkJettyDrainTimeouts();
+    void checkJettyDrainTimeouts(
+        std::unordered_map<volatile int*, int>& jetty_depth_set,
+        std::vector<Transport::Slice*>& failed_slices,
+        std::vector<UbEndPoint*>& deferred_deletes);
 
    private:
     int construct(GlobalConfig& config) override;
@@ -171,8 +175,20 @@ class UrmaContext : public UbContext {
 
 // define the UrmaEndpoint class
 class UrmaEndpoint : public UbEndPoint {
+    // UrmaContext::poll drives the jetty state machine via processWrCompletion.
+    friend class UrmaContext;
+
    public:
-    enum JettyState { ACTIVE = 0, DRAINING = 1, REBUILDING = 2 };
+    // PENDING_DRAIN: the slot hit ACK timeout while another jetty of this
+    // endpoint was draining/rebuilding. It is excluded from post selection
+    // and drained once the in-flight rebuild completes (rebuilds are
+    // serialized per endpoint).
+    enum JettyState {
+        ACTIVE = 0,
+        DRAINING = 1,
+        REBUILDING = 2,
+        PENDING_DRAIN = 3
+    };
 
     UrmaEndpoint(UrmaContext* context)
         : context_(context), jfc_outstanding_(nullptr) {}
@@ -198,9 +214,16 @@ class UrmaEndpoint : public UbEndPoint {
 
     // Called from UrmaContext::poll on ACK timeout / flush-done / drain
     // timeout.
-    void onJettyError(int slot);
-    void onFlushDone(int slot);
-    void checkDrainTimeout();
+    void onJettyError(int slot, std::vector<UbEndPoint*>& deferred_deletes);
+    void onFlushDone(int slot,
+                     std::unordered_map<volatile int*, int>& jetty_depth_set,
+                     std::vector<Transport::Slice*>& failed_slices,
+                     std::vector<UbEndPoint*>& deferred_deletes,
+                     int& resolved_wr_count);
+    void checkDrainTimeout(
+        std::unordered_map<volatile int*, int>& jetty_depth_set,
+        std::vector<Transport::Slice*>& failed_slices,
+        std::vector<UbEndPoint*>& deferred_deletes);
 
     int findSlotByDepth(volatile int* depth) const;
 
@@ -220,7 +243,31 @@ class UrmaEndpoint : public UbEndPoint {
 
     bool hasNonActiveJettyUnlocked() const;
     int selectActiveJettyUnlocked();
-    int rebuildJettyUnlocked(int slot);
+    // Transitions `slot` to DRAINING via urma_modify_jetty(ERROR) and arms
+    // the flush-done wait. Caller must hold lock_. Returns ERR_ENDPOINT if
+    // modify failed (caller should fall back to deleting the endpoint).
+    int startDrainUnlocked(int slot);
+    int rebuildJettyUnlocked(
+        int slot, std::unordered_map<volatile int*, int>& jetty_depth_set,
+        std::vector<Transport::Slice*>& failed_slices,
+        std::vector<UbEndPoint*>& deferred_deletes, int& resolved_wr_count);
+
+    // Delivers one WR completion to the normal success/failure path.
+    // Returns true when the completion resolved a live WR (and must be counted
+    // in the JFC outstanding accounting). Returns false for completions from a
+    // stale jetty generation, which were already resolved during the rebuild
+    // flush and must be dropped entirely. When allow_error_trigger is false
+    // (the caller already holds lock_ while rebuilding/draining), an
+    // ACK_TIMEOUT completion is delivered as a plain failure without
+    // re-entering onJettyError.
+    bool processWrCompletion(
+        urma_cr_t& cr, std::unordered_map<volatile int*, int>& jetty_depth_set,
+        std::vector<Transport::Slice*>& failed_slices,
+        std::vector<UbEndPoint*>& deferred_deletes, int jfc_index,
+        bool allow_error_trigger);
+
+    int recreateJettyUnlocked(int slot, urma_jfc_t* reuse_jfc,
+                              urma_jfr_t* reuse_jfr);
 
     static constexpr uint64_t kJettyDrainTimeoutNs = 3000000000ull;  // 3s
 
@@ -239,6 +286,7 @@ class UrmaEndpoint : public UbEndPoint {
     std::string peer_eid_;
     uint64_t drain_start_ns_ = 0;
     int draining_slot_ = -1;
+    std::vector<uint64_t> jetty_epoch_;
 };
 }  // namespace mooncake
 #endif  // URMA_ENDPOINT_H
diff --git a/mooncake-transfer-engine/include/transport/transport.h b/mooncake-transfer-engine/include/transport/transport.h
index c235bb3424..256d81e5a2 100644
--- a/mooncake-transfer-engine/include/transport/transport.h
+++ b/mooncake-transfer-engine/include/transport/transport.h
@@ -157,6 +157,7 @@ class Transport {
             struct {
                 uint64_t dest_addr;
                 volatile int *jetty_depth;
+                uint64_t jetty_epoch;
                 uint32_t retry_cnt;
                 uint32_t max_retry_cnt;
                 void *r_seg;
diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp
index 46a3d1e467..60e58fe2f0 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp
@@ -406,22 +406,26 @@ void UbWorkerPool::performPoll(int thread_id) {
     // are NOT returned from poll(), so this worker never dereferences them
     // after they may have been recycled by the submitting thread.
     std::unordered_map<volatile int*, int> jetty_depth_set;
+    std::vector<UbEndPoint*> deferred_deletes;
     std::vector<UbTransport::Slice*> failed_slices;
     for (int jfc_index = thread_id; jfc_index < context_.jfcCount();
          jfc_index += kTransferWorkerCount) {
-        UbTransport::Slice* failed[kPollCount];
-        int num_failed = 0;
-        int nr_poll = context_.poll(kPollCount, failed, num_failed,
-                                    jetty_depth_set, jfc_index);
-        if (nr_poll < 0) {
+        // poll() may also append completions recovered while draining a jetty
+        // (up to the full queue depth), so this must be a vector rather than
+        // a fixed-size array of kPollCount.
+        std::vector<UbTransport::Slice*> failed;
+        const size_t failed_before = failed.size();
+        int nr_resolved = context_.poll(kPollCount, failed, jetty_depth_set,
+                                        deferred_deletes, jfc_index);
+        if (nr_resolved < 0) {
             LOG(ERROR) << "Worker: Failed to poll jetty for complete";
             continue;
         }
-        int num_success = nr_poll - num_failed;
+        const size_t new_failed = failed.size() - failed_before;
+        const int num_success = nr_resolved - static_cast<int>(new_failed);
         success_nr_polls += num_success;
         processed_slice_count += num_success;
-        for (int i = 0; i < num_failed; ++i) {
-            UbTransport::Slice* slice = failed[i];
+        for (auto* slice : failed) {
             assert(slice);
             failed_nr_polls++;
             if (context_.active() && failed_nr_polls > 32 &&
@@ -439,13 +443,17 @@ void UbWorkerPool::performPoll(int thread_id) {
                 redispatch_counter_++;
             }
         }
-        if (nr_poll)
-            __sync_fetch_and_sub(context_.outstandingCount(jfc_index), nr_poll);
+        if (nr_resolved)
+            __sync_fetch_and_sub(context_.outstandingCount(jfc_index),
+                                 nr_resolved);
     }
 
     for (auto& entry : jetty_depth_set)
         __sync_fetch_and_sub(entry.first, entry.second);
 
+    for (auto* endpoint : deferred_deletes)
+        context_.deleteEndpointByPtr(endpoint);
+
     // Slices that hit max_retry: final markFailed() after all reads (and the
     // jetty depth returns above) are done. Failed slices were never published
     // by poll(), so they remained safe to deref up to this point.
diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
index 14a317f3d9..c63ae43540 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
@@ -549,7 +549,10 @@ void UrmaContext::removeDrainingEndpoint(UrmaEndpoint* endpoint) {
     draining_endpoints_.erase(endpoint);
 }
 
-void UrmaContext::checkJettyDrainTimeouts() {
+void UrmaContext::checkJettyDrainTimeouts(
+    std::unordered_map<volatile int*, int>& jetty_depth_set,
+    std::vector<Transport::Slice*>& failed_slices,
+    std::vector<UbEndPoint*>& deferred_deletes) {
     std::vector<UrmaEndpoint*> endpoints;
     {
         RWSpinlock::ReadGuard guard(jetty_owner_lock_);
@@ -557,16 +560,27 @@ void UrmaContext::checkJettyDrainTimeouts() {
                          draining_endpoints_.end());
     }
     for (auto* endpoint : endpoints) {
-        if (endpoint) endpoint->checkDrainTimeout();
+        if (endpoint)
+            endpoint->checkDrainTimeout(jetty_depth_set, failed_slices,
+                                        deferred_deletes);
     }
 }
 
-int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
-                      int& num_failed,
+namespace {
+void deferEndpointDelete(UrmaEndpoint* endpoint,
+                         std::vector<UbEndPoint*>& deferred_deletes) {
+    for (auto* ep : deferred_deletes) {
+        if (ep == endpoint) return;
+    }
+    deferred_deletes.push_back(endpoint);
+}
+}  // namespace
+
+int UrmaContext::poll(int num_entries,
+                      std::vector<Transport::Slice*>& failed_slices,
                       std::unordered_map<volatile int*, int>& jetty_depth_set,
+                      std::vector<UbEndPoint*>& deferred_deletes,
                       int jfc_index) {
-    num_failed = 0;
-    checkJettyDrainTimeouts();
     urma_cr_t cr[num_entries];
     int nr_poll = urma_poll_jfc(jfc_list_[jfc_index].native, num_entries, cr);
     if (nr_poll < 0) {
@@ -581,7 +595,8 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
             UrmaEndpoint* endpoint = nullptr;
             int slot = -1;
             if (findJettyOwner(cr[i].local_id, &endpoint, &slot) && endpoint) {
-                endpoint->onFlushDone(slot);
+                endpoint->onFlushDone(slot, jetty_depth_set, failed_slices,
+                                      deferred_deletes, wr_completions);
             } else {
                 LOG(WARNING) << "FLUSH_ERR_DONE for unknown jetty local_id="
                              << cr[i].local_id << " on " << device_name_;
@@ -593,65 +608,22 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices,
         if (!slice) {
             continue;
         }
-        ++wr_completions;
-
-        // All deref of `slice` (including the jetty_depth aggregation below)
-        // MUST happen before markSuccess(): once that publishes completion,
-        // the submitting thread may recycle the slice immediately.
-        auto* depth = slice->ub.jetty_depth;
-        auto it = jetty_depth_set.find(depth);
-        if (it != jetty_depth_set.end())
-            it->second++;
-        else
-            jetty_depth_set[depth] = 1;
-
-        if (cr[i].status == URMA_CR_SUCCESS) {
-            // Safe to publish here — we are done with this slice and do not
-            // return it to the caller, so no one else will deref it.
-            slice->markSuccess();
+        auto* endpoint = static_cast<UrmaEndpoint*>(slice->ub.endpoint);
+        if (!endpoint) {
+            // Keep the legacy accounting: the WR resolved even though we no
+            // longer know its endpoint.
+            ++wr_completions;
             continue;
         }
-
-        if (cr[i].status == URMA_CR_ACK_TIMEOUT_ERR) {
-            auto* endpoint = static_cast<UrmaEndpoint*>(slice->ub.endpoint);
-            if (endpoint) {
-                int slot = endpoint->findSlotByDepth(depth);
-                if (slot < 0) {
-                    UrmaEndpoint* mapped = nullptr;
-                    int mapped_slot = -1;
-                    if (findJettyOwner(cr[i].local_id, &mapped, &mapped_slot) &&
-                        mapped == endpoint) {
-                        slot = mapped_slot;
-                    }
-                }
-                if (slot >= 0) endpoint->onJettyError(slot);
-            }
+        if (endpoint->processWrCompletion(cr[i], jetty_depth_set, failed_slices,
+                                          deferred_deletes, jfc_index,
+                                          /*allow_error_trigger=*/true)) {
+            ++wr_completions;
         }
-
-        if (cr[i].status != URMA_CR_WR_FLUSH_ERR ||
-            show_work_request_flushed_error_)
-            LOG(ERROR) << "Worker: Process failed for slice (opcode: "
-                       << slice->opcode
-                       << ", source_addr: " << slice->source_addr
-                       << ", length: " << slice->length
-                       << ", dest_addr: " << (void*)slice->ub.dest_addr
-                       << ", local_nic: " << deviceName()
-                       << ", peer_nic: " << slice->peer_nic_path
-                       << ", dest_seg_tokenid: "
-                       << static_cast<urma_target_seg_t*>(slice->ub.r_seg)
-                              ->seg.token_id
-                       << ", retry_cnt: " << slice->ub.retry_cnt
-                       << "): " << cr[i].status << ", jfc idx : " << jfc_index
-                       << ", comp_events_acked: "
-                       << jfc_list_[jfc_index].native->comp_events_acked << " "
-                       << jfc_list_[jfc_index].native->async_events_acked;
-
-        // Failed: hand the slice back so the caller can decide retry vs
-        // final markFailed(). Slice is NOT published, so the caller may
-        // safely deref it.
-        failed_slices[num_failed++] = slice;
-    }
-    // Exclude FLUSH_ERR_DONE from outstanding accounting (it was never posted).
+    }
+    checkJettyDrainTimeouts(jetty_depth_set, failed_slices, deferred_deletes);
+    // Exclude FLUSH_ERR_DONE and dropped stale completions from outstanding
+    // accounting; the rebuild path already accounted for the stale ones.
     return wr_completions;
 }
 
@@ -703,6 +675,7 @@ int UrmaEndpoint::construct(GlobalConfig& config) {
 
     jetty_list_.resize(num_jetty_list);
     jetty_state_.assign(num_jetty_list, ACTIVE);
+    jetty_epoch_.assign(num_jetty_list, 1);
     peer_jetty_id_.assign(num_jetty_list, 0);
     jetty_id_map_.clear();
     peer_eid_.clear();
@@ -740,6 +713,19 @@ int UrmaEndpoint::construct(GlobalConfig& config) {
         jetty_list_[i] = urma_create_jetty(context_->urma_context_, &attr);
         if (!jetty_list_[i]) {
             PLOG(ERROR) << "Failed to create jetty";
+            for (size_t j = 0; j < i; ++j) {
+                if (!jetty_list_[j]) continue;
+                context_->unregisterJettyOwner(jetty_list_[j]->jetty_id.id);
+                urma_delete_jetty(jetty_list_[j]);
+                jetty_list_[j] = nullptr;
+            }
+            jetty_list_.clear();
+            jetty_state_.clear();
+            jetty_epoch_.clear();
+            peer_jetty_id_.clear();
+            jetty_id_map_.clear();
+            delete[] wr_depth_list_;
+            wr_depth_list_ = nullptr;
             return ERR_ENDPOINT;
         }
         uint32_t jetty_id = jetty_list_[i]->jetty_id.id;
@@ -896,10 +882,21 @@ void UrmaEndpoint::disconnectUnlocked() {
     draining_slot_ = -1;
     drain_start_ns_ = 0;
     for (size_t i = 0; i < jetty_list_.size(); ++i) {
-        int ret = urma_modify_jetty(jetty_list_[i], &attr);
-        if (ret) PLOG(ERROR) << "Failed to modify jetty to RESET";
+        if (!jetty_list_[i]) continue;
+        // Only jettys that entered ERROR (modify already called) are
+        // flushable; PENDING_DRAIN ones have not been modified yet and go
+        // through the normal RESET path below.
+        if (jetty_state_[i] == DRAINING || jetty_state_[i] == REBUILDING) {
+            urma_cr_t flush_crs[64];
+            while (true) {
+                int flushed = urma_flush_jetty(jetty_list_[i], 64, flush_crs);
+                if (flushed <= 0) break;
+            }
+        }
+        int reset_ret = urma_modify_jetty(jetty_list_[i], &attr);
+        if (reset_ret) PLOG(ERROR) << "Failed to modify jetty to RESET";
         auto imported_jetty = imported_jetty_map_[jetty_list_[i]];
-        ret = urma_unbind_jetty(jetty_list_[i]);
+        int ret = urma_unbind_jetty(jetty_list_[i]);
         if (ret) PLOG(ERROR) << "Failed to unbind jetty";
         ret = urma_unimport_jetty(imported_jetty);
         if (ret) PLOG(ERROR) << "Failed to unimport jetty";
@@ -914,7 +911,7 @@ void UrmaEndpoint::disconnectUnlocked() {
             __sync_fetch_and_sub(jfc_outstanding_, wr_depth_list_[i]);
             wr_depth_list_[i] = 0;
         }
-        jetty_state_[i] = ACTIVE;
+        if (!reset_ret) jetty_state_[i] = ACTIVE;
     }
     imported_jetty_map_.clear();
     peer_jetty_id_.assign(jetty_list_.size(), 0);
@@ -1031,6 +1028,7 @@ int UrmaEndpoint::submitPostSend(
         slice->ts = getCurrentTimeInNano();
         slice->status = Transport::Slice::POSTED;
         slice->ub.jetty_depth = &wr_depth_list_[jetty_index];
+        slice->ub.jetty_epoch = jetty_epoch_[jetty_index];
         // Set endpoint pointer for each slice before submitting
         slice->ub.endpoint = this;
     }
@@ -1151,53 +1149,62 @@ int UrmaEndpoint::selectActiveJettyUnlocked() {
     return -1;
 }
 
-void UrmaEndpoint::onJettyError(int slot) {
+int UrmaEndpoint::startDrainUnlocked(int slot) {
+    urma_jetty_attr_t attr{};
+    attr.mask = JETTY_STATE;
+    attr.state = URMA_JETTY_STATE_ERROR;
+    int ret = urma_modify_jetty(jetty_list_[slot], &attr);
+    if (ret) {
+        PLOG(ERROR) << "Failed to modify jetty to ERROR, slot=" << slot
+                    << " jetty_id=" << jetty_list_[slot]->jetty_id.id;
+        context_->removeDrainingEndpoint(this);
+        draining_slot_ = -1;
+        drain_start_ns_ = 0;
+        return ERR_ENDPOINT;
+    }
+    jetty_state_[slot] = DRAINING;
+    draining_slot_ = slot;
+    drain_start_ns_ = getCurrentTimeInNano();
+    context_->addDrainingEndpoint(this);
+    LOG(WARNING) << "Jetty ACK timeout: start drain slot=" << slot
+                 << " jetty_id=" << jetty_list_[slot]->jetty_id.id << " on "
+                 << toString();
+    return 0;
+}
+
+void UrmaEndpoint::onJettyError(int slot,
+                                std::vector<UbEndPoint*>& deferred_deletes) {
     bool delete_ep = false;
     {
         RWSpinlock::WriteGuard guard(lock_);
         if (slot < 0 || slot >= static_cast<int>(jetty_list_.size())) return;
-        if (jetty_state_[slot] == DRAINING ||
-            jetty_state_[slot] == REBUILDING) {
-            return;  // idempotent
-        }
-        // Serial rebuild: at most one non-ACTIVE jetty per endpoint.
-        if (hasNonActiveJettyUnlocked()) {
-            LOG(INFO) << "Skip jetty rebuild for slot " << slot
-                      << ": another jetty is already draining/rebuilding on "
-                      << toString();
-            return;
-        }
+        if (jetty_state_[slot] != ACTIVE) return;  // idempotent
         if (!jetty_list_[slot]) return;
-
-        urma_jetty_attr_t attr{};
-        attr.mask = JETTY_STATE;
-        attr.state = URMA_JETTY_STATE_ERROR;
-        int ret = urma_modify_jetty(jetty_list_[slot], &attr);
-        if (ret) {
-            PLOG(ERROR) << "Failed to modify jetty to ERROR, slot=" << slot
-                        << " jetty_id=" << jetty_list_[slot]->jetty_id.id;
-            context_->removeDrainingEndpoint(this);
-            draining_slot_ = -1;
-            drain_start_ns_ = 0;
-            delete_ep = true;
-        } else {
-            jetty_state_[slot] = DRAINING;
-            draining_slot_ = slot;
-            drain_start_ns_ = getCurrentTimeInNano();
-            context_->addDrainingEndpoint(this);
-            LOG(WARNING) << "Jetty ACK timeout: start drain slot=" << slot
+        // Serial rebuild: at most one draining/rebuilding jetty per endpoint.
+        // Queue this slot as PENDING_DRAIN so that submitPostSend stops
+        // selecting it; the rebuild tail starts its drain once serialized
+        // state is free again.
+        if (hasNonActiveJettyUnlocked()) {
+            jetty_state_[slot] = PENDING_DRAIN;
+            LOG(WARNING) << "Queue jetty drain for slot " << slot
                          << " jetty_id=" << jetty_list_[slot]->jetty_id.id
-                         << " on " << toString();
+                         << ": another jetty is already draining/rebuilding on "
+                         << toString();
+            return;
         }
+        if (startDrainUnlocked(slot)) delete_ep = true;
     }
     if (delete_ep) {
         LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
                    << "modify_jetty(ERROR) failed on " << toString();
-        context_->deleteEndpointByPtr(this);
+        deferEndpointDelete(this, deferred_deletes);
     }
 }
 
-void UrmaEndpoint::onFlushDone(int slot) {
+void UrmaEndpoint::onFlushDone(
+    int slot, std::unordered_map<volatile int*, int>& jetty_depth_set,
+    std::vector<Transport::Slice*>& failed_slices,
+    std::vector<UbEndPoint*>& deferred_deletes, int& resolved_wr_count) {
     bool delete_ep = false;
     {
         RWSpinlock::WriteGuard guard(lock_);
@@ -1206,7 +1213,8 @@ void UrmaEndpoint::onFlushDone(int slot) {
         jetty_state_[slot] = REBUILDING;
         LOG(INFO) << "Jetty flush-done: rebuild slot=" << slot << " on "
                   << toString();
-        if (rebuildJettyUnlocked(slot)) {
+        if (rebuildJettyUnlocked(slot, jetty_depth_set, failed_slices,
+                                 deferred_deletes, resolved_wr_count)) {
             context_->removeDrainingEndpoint(this);
             draining_slot_ = -1;
             drain_start_ns_ = 0;
@@ -1216,11 +1224,14 @@ void UrmaEndpoint::onFlushDone(int slot) {
     if (delete_ep) {
         LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
                    << "rebuildJetty failed on " << toString();
-        context_->deleteEndpointByPtr(this);
+        deferEndpointDelete(this, deferred_deletes);
     }
 }
 
-void UrmaEndpoint::checkDrainTimeout() {
+void UrmaEndpoint::checkDrainTimeout(
+    std::unordered_map<volatile int*, int>& /*jetty_depth_set*/,
+    std::vector<Transport::Slice*>& /*failed_slices*/,
+    std::vector<UbEndPoint*>& deferred_deletes) {
     bool delete_ep = false;
     {
         RWSpinlock::WriteGuard guard(lock_);
@@ -1243,11 +1254,75 @@ void UrmaEndpoint::checkDrainTimeout() {
     if (delete_ep) {
         LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: "
                    << "flush-done timeout on " << toString();
-        context_->deleteEndpointByPtr(this);
+        deferEndpointDelete(this, deferred_deletes);
     }
 }
 
-int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
+bool UrmaEndpoint::processWrCompletion(
+    urma_cr_t& cr, std::unordered_map<volatile int*, int>& jetty_depth_set,
+    std::vector<Transport::Slice*>& failed_slices,
+    std::vector<UbEndPoint*>& deferred_deletes, int jfc_index,
+    bool allow_error_trigger) {
+    auto slice = reinterpret_cast<Transport::Slice*>(cr.user_ctx);
+    if (!slice) return false;
+
+    // All deref of `slice` (including jetty_depth aggregation below) MUST
+    // happen before markSuccess(): once that publishes completion, the
+    // submitting thread may recycle the slice immediately.
+    auto* depth = slice->ub.jetty_depth;
+    int slot = findSlotByDepth(depth);
+    if (slot >= 0 && slot < static_cast<int>(jetty_epoch_.size()) &&
+        slice->ub.jetty_epoch != jetty_epoch_[slot]) {
+        LOG(WARNING) << "Dropping stale jetty completion for slot " << slot
+                     << " on " << toString();
+        return false;
+    }
+
+    auto it = jetty_depth_set.find(depth);
+    if (it != jetty_depth_set.end())
+        it->second++;
+    else
+        jetty_depth_set[depth] = 1;
+
+    if (cr.status == URMA_CR_SUCCESS) {
+        slice->markSuccess();
+        return true;
+    }
+
+    if (cr.status == URMA_CR_ACK_TIMEOUT_ERR && allow_error_trigger) {
+        if (slot < 0) {
+            UrmaEndpoint* mapped = nullptr;
+            int mapped_slot = -1;
+            if (context_->findJettyOwner(cr.local_id, &mapped, &mapped_slot) &&
+                mapped == this) {
+                slot = mapped_slot;
+            }
+        }
+        if (slot >= 0) onJettyError(slot, deferred_deletes);
+    }
+
+    if (cr.status != URMA_CR_WR_FLUSH_ERR ||
+        context_->traceWorkRequestFlushedErrors())
+        LOG(ERROR) << "Worker: Process failed for slice (opcode: "
+                   << slice->opcode << ", source_addr: " << slice->source_addr
+                   << ", length: " << slice->length
+                   << ", dest_addr: " << (void*)slice->ub.dest_addr
+                   << ", local_nic: " << context_->deviceName()
+                   << ", peer_nic: " << slice->peer_nic_path
+                   << ", dest_seg_tokenid: "
+                   << static_cast<urma_target_seg_t*>(slice->ub.r_seg)
+                          ->seg.token_id
+                   << ", retry_cnt: " << slice->ub.retry_cnt
+                   << "): " << cr.status << ", jfc idx : " << jfc_index;
+
+    failed_slices.push_back(slice);
+    return true;
+}
+
+int UrmaEndpoint::rebuildJettyUnlocked(
+    int slot, std::unordered_map<volatile int*, int>& jetty_depth_set,
+    std::vector<Transport::Slice*>& failed_slices,
+    std::vector<UbEndPoint*>& deferred_deletes, int& resolved_wr_count) {
     auto* old_jetty = jetty_list_[slot];
     if (!old_jetty) return ERR_ENDPOINT;
     const uint32_t old_id = old_jetty->jetty_id.id;
@@ -1256,7 +1331,7 @@ int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
     urma_jfc_t* reuse_jfc = old_jetty->jetty_cfg.jfs_cfg.jfc;
     urma_jfr_t* reuse_jfr = old_jetty->jetty_cfg.shared.jfr;
 
-    // 1) Flush residual WRs (may overlap with already-polled CRs).
+    // 1) Flush residual WRs and deliver their completions.
     urma_cr_t flush_crs[64];
     while (true) {
         int flushed = urma_flush_jetty(old_jetty, 64, flush_crs);
@@ -1265,8 +1340,14 @@ int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
             return ERR_ENDPOINT;
         }
         if (flushed == 0) break;
-        // Completions for these WRs should already have been (or will be)
-        // accounted via poll; do not touch slice pointers from flush CRs.
+        for (int j = 0; j < flushed; ++j) {
+            if (flush_crs[j].status == URMA_CR_WR_FLUSH_ERR_DONE) continue;
+            if (processWrCompletion(flush_crs[j], jetty_depth_set, failed_slices,
+                                    deferred_deletes, -1,
+                                    /*allow_error_trigger=*/false)) {
+                ++resolved_wr_count;
+            }
+        }
     }
 
     // 2) Unbind / unimport old peer view.
@@ -1292,44 +1373,36 @@ int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
         return ERR_ENDPOINT;
     }
     jetty_list_[slot] = nullptr;
-    if (wr_depth_list_[slot] != 0) {
-        __sync_fetch_and_sub(jfc_outstanding_, wr_depth_list_[slot]);
-        wr_depth_list_[slot] = 0;
-    }
-
-    // 4) Create replacement jetty with the same JFC/JFR config.
-    urma_jfs_cfg_t jfs_cfg = {
-        .depth = 2048,
-        .trans_mode = URMA_TM_RC,
-        .priority = 15,
-        .max_sge = 5,
-        .rnr_retry = 7,
-        .err_timeout = 17,
-        .user_ctx = 0,
-    };
-    urma_jetty_flag_t jetty_flag = {};
-    jetty_flag.bs.share_jfr = 1;
-    urma_jetty_cfg_t attr{};
-    attr.flag = jetty_flag;
-    attr.jfs_cfg = jfs_cfg;
-    attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc();
-    attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr();
-    urma_jetty_t* new_jetty = urma_create_jetty(context_->urma_context_, &attr);
-    if (!new_jetty) {
-        PLOG(ERROR) << "Failed to create jetty during rebuild";
+    // No explicit depth/outstanding adjustment here: every WR of the old
+    // jetty completes exactly once, either via JFC poll (WR_FLUSH_ERR) before
+    // the flush-done fence or via the urma_flush_jetty loop above, and each
+    // completion was already accounted through processWrCompletion (deferred
+    // via jetty_depth_set / resolved_wr_count). Adjusting wr_depth_list_[slot]
+    // here would double-count. The epoch bump makes any late duplicate
+    // completion for the old generation a no-op instead.
+    ++jetty_epoch_[slot];
+
+    if (recreateJettyUnlocked(slot, reuse_jfc, reuse_jfr)) {
         return ERR_ENDPOINT;
     }
+    urma_jetty_t* new_jetty = jetty_list_[slot];
 
     // 5) Re-import peer and bind locally (no peer protocol).
     if (peer_eid_.empty()) {
         LOG(ERROR) << "Missing peer eid during jetty rebuild";
+        context_->unregisterJettyOwner(new_jetty->jetty_id.id);
+        jetty_id_map_.erase(new_jetty->jetty_id.id);
         urma_delete_jetty(new_jetty);
+        jetty_list_[slot] = nullptr;
         return ERR_ENDPOINT;
     }
     urma_eid_t eid;
     if (!context_->transEidFromString(peer_eid_, eid)) {
         LOG(ERROR) << "Invalid peer eid during jetty rebuild: " << peer_eid_;
+        context_->unregisterJettyOwner(new_jetty->jetty_id.id);
+        jetty_id_map_.erase(new_jetty->jetty_id.id);
         urma_delete_jetty(new_jetty);
+        jetty_list_[slot] = nullptr;
         return ERR_ENDPOINT;
     }
     urma_rjetty_t rjetty = {};
@@ -1343,32 +1416,78 @@ int UrmaEndpoint::rebuildJettyUnlocked(int slot) {
         urma_import_jetty(context_->urma_context_, &rjetty, &urma_token);
     if (!imported) {
         PLOG(ERROR) << "Failed to import peer jetty during rebuild";
+        context_->unregisterJettyOwner(new_jetty->jetty_id.id);
+        jetty_id_map_.erase(new_jetty->jetty_id.id);
         urma_delete_jetty(new_jetty);
+        jetty_list_[slot] = nullptr;
         return ERR_ENDPOINT;
     }
     urma_status_t bind_ret = urma_bind_jetty(new_jetty, imported);
     if (bind_ret != URMA_SUCCESS && bind_ret != URMA_EEXIST) {
         PLOG(ERROR) << "Failed to bind rebuilt jetty";
         urma_unimport_jetty(imported);
+        context_->unregisterJettyOwner(new_jetty->jetty_id.id);
+        jetty_id_map_.erase(new_jetty->jetty_id.id);
         urma_delete_jetty(new_jetty);
+        jetty_list_[slot] = nullptr;
         return ERR_ENDPOINT;
     }
 
-    jetty_list_[slot] = new_jetty;
     imported_jetty_map_[new_jetty] = imported;
-    const uint32_t new_id = new_jetty->jetty_id.id;
-    jetty_id_map_[new_id] = slot;
-    context_->registerJettyOwner(new_id, this, slot);
     jetty_state_[slot] = ACTIVE;
     draining_slot_ = -1;
     drain_start_ns_ = 0;
     context_->removeDrainingEndpoint(this);
 
     LOG(WARNING) << "Jetty rebuilt successfully slot=" << slot
-                 << " old_id=" << old_id << " new_id=" << new_id
+                 << " old_id=" << old_id << " new_id=" << new_jetty->jetty_id.id
                  << " peer_id=" << peer_id << " elapsed_ms="
                  << ((getCurrentTimeInNano() - started_ns) / 1000000ull)
                  << " on " << toString();
+
+    // Another jetty may have hit ACK timeout while this one was rebuilding;
+    // start draining the queued slot now (still one at a time).
+    for (size_t i = 0; i < jetty_state_.size(); ++i) {
+        if (jetty_state_[i] != PENDING_DRAIN || !jetty_list_[i]) continue;
+        if (startDrainUnlocked(static_cast<int>(i))) {
+            LOG(ERROR) << "Failed to start drain for queued slot " << i
+                       << " on " << toString();
+            return ERR_ENDPOINT;
+        }
+        break;
+    }
+    return 0;
+}
+
+int UrmaEndpoint::recreateJettyUnlocked(int slot, urma_jfc_t* reuse_jfc,
+                                        urma_jfr_t* reuse_jfr) {
+    urma_jfs_cfg_t jfs_cfg = {
+        .depth = 2048,
+        .trans_mode = URMA_TM_RC,
+        .priority = 15,
+        .max_sge = 5,
+        .rnr_retry = 7,
+        .err_timeout = 17,
+        .user_ctx = 0,
+    };
+    urma_jetty_flag_t jetty_flag = {};
+    jetty_flag.bs.share_jfr = 1;
+    urma_jetty_cfg_t attr{};
+    attr.flag = jetty_flag;
+    attr.jfs_cfg = jfs_cfg;
+    attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc();
+    attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr();
+    urma_jetty_t* new_jetty =
+        urma_create_jetty(context_->urma_context_, &attr);
+    if (!new_jetty) {
+        PLOG(ERROR) << "Failed to create jetty during rebuild";
+        return ERR_ENDPOINT;
+    }
+
+    jetty_list_[slot] = new_jetty;
+    const uint32_t new_id = new_jetty->jetty_id.id;
+    jetty_id_map_[new_id] = slot;
+    context_->registerJettyOwner(new_id, this, slot);
     return 0;
 }
 

From 1112d30f63c7ddade6fee85e264198f3bc62aa00 Mon Sep 17 00:00:00 2001
From: Connor-Matthew <60215777+Connor-Matthew@users.noreply.github.com>
Date: Thu, 27 Aug 2026 00:47:41 +0800
Subject: [PATCH 4/4] [TransferEngine] Apply clang-format to jetty rebuild
 safety fixes

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../kunpeng_transport/urma/urma_endpoint.cpp  | 28 +++++++++----------
 1 file changed, 13 insertions(+), 15 deletions(-)

diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
index c63ae43540..3a7e18e401 100644
--- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
+++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp
@@ -1303,17 +1303,16 @@ bool UrmaEndpoint::processWrCompletion(
 
     if (cr.status != URMA_CR_WR_FLUSH_ERR ||
         context_->traceWorkRequestFlushedErrors())
-        LOG(ERROR) << "Worker: Process failed for slice (opcode: "
-                   << slice->opcode << ", source_addr: " << slice->source_addr
-                   << ", length: " << slice->length
-                   << ", dest_addr: " << (void*)slice->ub.dest_addr
-                   << ", local_nic: " << context_->deviceName()
-                   << ", peer_nic: " << slice->peer_nic_path
-                   << ", dest_seg_tokenid: "
-                   << static_cast<urma_target_seg_t*>(slice->ub.r_seg)
-                          ->seg.token_id
-                   << ", retry_cnt: " << slice->ub.retry_cnt
-                   << "): " << cr.status << ", jfc idx : " << jfc_index;
+        LOG(ERROR)
+            << "Worker: Process failed for slice (opcode: " << slice->opcode
+            << ", source_addr: " << slice->source_addr
+            << ", length: " << slice->length
+            << ", dest_addr: " << (void*)slice->ub.dest_addr
+            << ", local_nic: " << context_->deviceName()
+            << ", peer_nic: " << slice->peer_nic_path << ", dest_seg_tokenid: "
+            << static_cast<urma_target_seg_t*>(slice->ub.r_seg)->seg.token_id
+            << ", retry_cnt: " << slice->ub.retry_cnt << "): " << cr.status
+            << ", jfc idx : " << jfc_index;
 
     failed_slices.push_back(slice);
     return true;
@@ -1342,8 +1341,8 @@ int UrmaEndpoint::rebuildJettyUnlocked(
         if (flushed == 0) break;
         for (int j = 0; j < flushed; ++j) {
             if (flush_crs[j].status == URMA_CR_WR_FLUSH_ERR_DONE) continue;
-            if (processWrCompletion(flush_crs[j], jetty_depth_set, failed_slices,
-                                    deferred_deletes, -1,
+            if (processWrCompletion(flush_crs[j], jetty_depth_set,
+                                    failed_slices, deferred_deletes, -1,
                                     /*allow_error_trigger=*/false)) {
                 ++resolved_wr_count;
             }
@@ -1477,8 +1476,7 @@ int UrmaEndpoint::recreateJettyUnlocked(int slot, urma_jfc_t* reuse_jfc,
     attr.jfs_cfg = jfs_cfg;
     attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc();
     attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr();
-    urma_jetty_t* new_jetty =
-        urma_create_jetty(context_->urma_context_, &attr);
+    urma_jetty_t* new_jetty = urma_create_jetty(context_->urma_context_, &attr);
     if (!new_jetty) {
         PLOG(ERROR) << "Failed to create jetty during rebuild";
         return ERR_ENDPOINT;