diff --git a/docs/cn/ubring.md b/docs/cn/ubring.md
index 1db9909103..cd100db84d 100644
--- a/docs/cn/ubring.md
+++ b/docs/cn/ubring.md
@@ -187,7 +187,7 @@ UBRing 架构包含以下组件：
 
 ### 定时器管理
 
-UBRing 使用高精度定时器系统 (`timer_mgr.cpp`) 进行连接管理和超时处理，支持 epoll（Linux）和 kqueue（macOS）。
+UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
 
 ## 参考资料
 
diff --git a/docs/en/ubring.md b/docs/en/ubring.md
index fbf87fff36..0cb3d08954 100644
--- a/docs/en/ubring.md
+++ b/docs/en/ubring.md
@@ -186,7 +186,7 @@ The shared memory manager (`shm_mgr.cpp`) provides a unified interface for diffe
 
 ### Timer Management
 
-UBRing uses a high-precision timer system (`timer_mgr.cpp`) for connection management and timeout handling, supporting both epoll (Linux) and kqueue (macOS).
+UBRing connection management and timeout handling are built on bthread timers (`bthread_timer_add`/`bthread_timer_del`, wrapped in `timer_mgr.cpp`). The non-blocking `UbrTimerDel` is safe to call from within a timer callback; external teardown that frees resources reachable from the callback argument uses `UbrTimerDelAndWait` to wait out a possibly running callback; one-shot timers clean up their handle automatically when they fire. Timer callbacks run on the process-wide bthread timer thread and must return quickly. The close-check timer backs off exponentially while the link is idle (capped by `ub_event_queue_timer_interval_max_us`, 10ms by default) and returns to fast polling once there is traffic or a close in progress.
 
 ## References
 
diff --git a/src/brpc/ubshm/common/common.h b/src/brpc/ubshm/common/common.h
index c64ae9843f..504da07c28 100644
--- a/src/brpc/ubshm/common/common.h
+++ b/src/brpc/ubshm/common/common.h
@@ -116,6 +116,7 @@ static inline int Copy64Byte(int8_t *dst, int8_t *src) {
 
 #define SEC_TO_NSEC 1000000000
 #define MSEC_TO_NSEC 1000000
+#define SEC_TO_USEC 1000000
 #define USEC_TO_NSEC 1000
 #define MSEC_TO_SEC 1000
 #define MAX_IP_PORT_STR_LEN 23
diff --git a/src/brpc/ubshm/shm/shm_ubs.cpp b/src/brpc/ubshm/shm/shm_ubs.cpp
index ccbd325923..a4f27ee1d8 100644
--- a/src/brpc/ubshm/shm/shm_ubs.cpp
+++ b/src/brpc/ubshm/shm/shm_ubs.cpp
@@ -47,7 +47,7 @@ DEFINE_int32(ub_flying_io_timeout_s, 5,
              "Time in seconds to wait for stopping data sending and receiving "
              "when the link is disconnected.");
 char g_region_name[MAX_REGION_NAME_DESC_LENGTH] = {0};
-int g_shm_timer_fd = 0;
+UbrTimerId g_shm_timer_id = nullptr;
 ShmList *g_shm_list = nullptr;
 static RETURN_CODE UbsShmInterfacesLoad(void);
 char hostname[MAX_HOST_NAME_DESC_LENGTH];
@@ -377,14 +377,15 @@ RETURN_CODE UbsShmInit(void)
 
 RETURN_CODE UbsShmFini(void)
 {
-    int ret = ubsmem_finalize();
-    if (ret != UBSM_OK) {
-        LOG(ERROR) << "Ubs shm finalize fail, ret=" << ret;
+    // Stop the cleanup timer before finalizing the SDK it calls into.
+    if (UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
+        LOG(ERROR) << "Ubs shm list finalize failed.";
         return UBRING_ERR;
     }
 
-    if (UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
-        LOG(ERROR) << "Ubs shm list finalize failed.";
+    int ret = ubsmem_finalize();
+    if (ret != UBSM_OK) {
+        LOG(ERROR) << "Ubs shm finalize fail, ret=" << ret;
         return UBRING_ERR;
     }
 
@@ -418,50 +419,56 @@ void *UbsShmCallback(void* args)
         return nullptr;
     }
 
-    LOCK_GUARD(shm_list->shm_lock);
-    while (shm_list->head != nullptr) {
-        SHM shm = shm_list->head->shm;
-        if (shm.addr == nullptr) {
-            LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+    // Drain one node per fire and keep the SDK calls outside the lock, so
+    // a slow daemon cannot stall the timer thread for the whole list.
+    SHM shm;
+    {
+        LOCK_GUARD(shm_list->shm_lock);
+        if (shm_list->head == nullptr) {
             return nullptr;
         }
+        shm = shm_list->head->shm;
+    }
+    if (shm.addr == nullptr) {
+        LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+        LOCK_GUARD(shm_list->shm_lock);
+        DeleteShmToList(shm_list);
+        return nullptr;
+    }
 
-        int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
-        if (ret != UBSM_OK) {
-            if (ret == UBSM_ERR_NET) {
-                return nullptr;
-            }
-            LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
-            return nullptr;
+    int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
+    if (ret != UBSM_OK) {
+        if (ret == UBSM_ERR_NET) {
+            return nullptr;              // retried on the next fire
         }
-        LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " success.";
+        LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
+        return nullptr;                  // node stays at head, retried
+    }
+    LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " success.";
 
-        ret = ubsmem_shmem_deallocate(shm.name);
-        if (ret != UBSM_OK) {
-            DeleteShmToList(shm_list);
-            LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
-            return nullptr;
-        }
+    {
+        LOCK_GUARD(shm_list->shm_lock);
         DeleteShmToList(shm_list);
-        LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len << " success.";
     }
 
+    ret = ubsmem_shmem_deallocate(shm.name);
+    if (ret != UBSM_OK) {
+        LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
+        return nullptr;
+    }
+    LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len << " success.";
     return nullptr;
 }
 
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
 {
-    const uint32_t timer_interval_s = FLAGS_ub_flying_io_timeout_s;
-    itimerspec time_spec = {
-        .it_interval = {.tv_sec = timer_interval_s, .tv_nsec = 0},
-        .it_value = {.tv_sec = 0, .tv_nsec = 1}
-    };
-    int timer_fd = TimerStart(&time_spec, UbsShmCallback, (void*)shm_list);
-    if (UNLIKELY(timer_fd == -1)) {
+    const uint64_t timer_interval_us = (uint64_t)FLAGS_ub_flying_io_timeout_s * SEC_TO_USEC;
+    RETURN_CODE rc = UbrTimerStart(&g_shm_timer_id, 0, timer_interval_us,
+                                   UbsShmCallback, (void*)shm_list);
+    if (UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start shm timer failed.";
         return UBRING_ERR;
     }
-    g_shm_timer_fd = timer_fd;
 
     return UBRING_OK;
 }
@@ -493,7 +500,8 @@ RETURN_CODE InitShmTimer(ShmList **shm_list)
 
 RETURN_CODE DestroyShmTimer(ShmList *shm_list)
 {
-    DeleteTimerSafe((uint32_t)g_shm_timer_fd);
+    // Wait out a possibly running UbsShmCallback before tearing shm_list down.
+    UbrTimerDelAndWait(&g_shm_timer_id);
     if (shm_list == nullptr) {
         LOG(WARNING) << "Shm list is null.";
         return UBRING_ERR;
diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp
index b5e0c9ef3b..d4d4e6d5de 100644
--- a/src/brpc/ubshm/timer/timer_mgr.cpp
+++ b/src/brpc/ubshm/timer/timer_mgr.cpp
@@ -15,454 +15,261 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#define _GNU_SOURCE
-#include <pthread.h>
-#include <sched.h>
-#include <errno.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <unistd.h>
 #include <atomic>
-#include <sys/resource.h>
+#include <new>
+#include "bthread/bthread.h"                     // bthread_usleep
+#include "bthread/unstable.h"                    // bthread_timer_add/del
+#include "butil/time.h"
 #include "brpc/ubshm/timer/timer_mgr.h"
 
 namespace brpc {
 namespace ubring {
 
-int32_t g_epoll_fd = -1;
-std::atomic<uint32_t> g_total_timer_num(0);
-TimerFdCtx *g_timer_fd_ctx_map = nullptr;
-uint32_t g_max_system_fd = 0;
-static pthread_t g_epoll_execute_thread = 0;
-static int32_t g_timer_module_initialized = 0;
-
-#if defined(OS_MACOSX)
-static int timerfd_create_macosx(int clockid, int flags);
-static int timerfd_settime_macosx(int fd, int flags,
-                                   const itimerspec *new_value,
-                                   itimerspec *old_value);
-#endif
-
-static RETURN_CODE DeleteTimerInner(uint32_t fd) {
-    if (g_timer_fd_ctx_map == nullptr) {
-        return UBRING_OK;
-    }
-
-    if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) {
-        return UBRING_ERR;
-    }
-
-    if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) {
-        pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock);
-        return UBRING_OK;
-    }
-
-    g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING;
-    g_timer_fd_ctx_map[fd].cb = nullptr;
-    g_timer_fd_ctx_map[fd].args = nullptr;
-    g_timer_fd_ctx_map[fd].periodical = 0;
-    g_timer_fd_ctx_map[fd].fd = 0;
-
-    pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock);
-
-#if defined(OS_LINUX)
-    epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr);
-#elif defined(OS_MACOSX)
-    struct kevent evt;
-    EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr);
-    kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr);
-#endif
-
-    uint64_t exp = 0;
-    read((int)fd, &exp, sizeof(exp));
-
-    close((int)fd);
-    std::atomic_fetch_sub(&g_total_timer_num, 1U);
-    return UBRING_OK;
-}
-
-static RETURN_CODE StartTimeEpoll(void) {
-#if defined(OS_LINUX)
-    g_epoll_fd = epoll_create1(0);
-#elif defined(OS_MACOSX)
-    g_epoll_fd = kqueue();
-#endif
-    if (UNLIKELY(g_epoll_fd == -1)) {
-        LOG(ERROR) << "Failed to create epoll/kqueue. errno=" << errno;
-        return UBRING_ERR;
-    }
-
-    int ret = pthread_create(&g_epoll_execute_thread, nullptr, TimerEpoll, nullptr);
-    if (UNLIKELY(ret != 0)) {
-        LOG(ERROR) << "Failed to create thread err=" << ret;
-        return UBRING_ERR;
-    }
-    return UBRING_OK;
-}
-
-static RETURN_CODE TimerSpinLocksInit(void) {
-    if (g_timer_fd_ctx_map == nullptr) {
-        LOG(ERROR) << "Timer module is not fully initialized.";
-        return UBRING_ERR;
-    }
-
-    for (uint32_t fd = 0; fd < g_max_system_fd; fd++) {
-        int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spin_lock,
-                                    PTHREAD_PROCESS_PRIVATE);
-        if (ret != EOK) {
-            LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd;
-            for (uint32_t cleanup_fd = 0; cleanup_fd < fd; cleanup_fd++) {
-                pthread_spin_destroy(&g_timer_fd_ctx_map[cleanup_fd].spin_lock);
-            }
-            return UBRING_ERR;
-        }
-    }
-    return UBRING_OK;
-}
-
-static RETURN_CODE ExecuteCallback(int32_t timer_fd) {
-    UnifiedCallback((void *)(&g_timer_fd_ctx_map[timer_fd]));
-    return UBRING_OK;
-}
-
-static RETURN_CODE TimerCtxMapCompletion(void) {
-    memset(g_timer_fd_ctx_map, 0, sizeof(TimerFdCtx) * g_max_system_fd);
-
-    RETURN_CODE ret = TimerSpinLocksInit();
-    if (ret != UBRING_OK) {
-        LOG(ERROR) << "Failed to init spin locks for timer module.";
-        return UBRING_ERR;
-    }
-    return UBRING_OK;
-}
-
-RETURN_CODE TimerInit(void) {
-    if (g_timer_module_initialized > 0) {
-        return UBRING_OK;
-    }
-
-    g_total_timer_num.store(0);
-
-    struct rlimit rlim;
-    if (getrlimit(RLIMIT_NOFILE, &rlim) != UBRING_OK) {
-        LOG(ERROR) << "Failed to get fd";
-        return UBRING_ERR;
-    }
-    g_max_system_fd = (uint32_t)rlim.rlim_cur;
-
-    if (g_timer_fd_ctx_map == nullptr) {
-        g_timer_fd_ctx_map = (TimerFdCtx *)malloc(sizeof(TimerFdCtx) * g_max_system_fd);
-        if (UNLIKELY(!g_timer_fd_ctx_map)) {
-            LOG(ERROR) << "Fail to malloc space for timer modules. errno=%d", errno;
-            return UBRING_ERR;
-        }
-
-        RETURN_CODE ret = TimerCtxMapCompletion();
-        if (ret != UBRING_OK) {
-            LOG(ERROR) << "Failed to init main data structure of Time Module. ret=" << ret;
-            free(g_timer_fd_ctx_map);
-            g_timer_fd_ctx_map = nullptr;
-            return UBRING_ERR;
-        }
-    }
-
-    RETURN_CODE ret = StartTimeEpoll();
-    if (ret != UBRING_OK) {
-        LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret;
-        if (LIKELY(g_timer_fd_ctx_map != nullptr)) {
-            FREE_PTR(g_timer_fd_ctx_map);
+namespace {
+
+enum UbrTimerState {
+    kStarting = 0,                               // published, not scheduled yet
+    kScheduled = 1,
+    kDead = 2                                    // scheduling failed
+};
+
+}  // namespace
+
+// Reference rules: one "owner" ref for the handle slot, one "schedule" ref
+// per pending/running bthread schedule, plus one ref held by the starter
+// until its post-schedule bookkeeping is done. The schedule ref is
+// consumed by the firing callback or by the deleter whose
+// bthread_timer_del returned 0 (cancelled before run); the owner ref is
+// consumed by whoever takes the task out of *slot -- a deleter, or the
+// one-shot firing callback itself, which exits the slot BEFORE running
+// the callback so that the callback may free the object storing the slot.
+// All atomics are seq_cst so no interleaving can release a ref twice or
+// free the task while a callback or the starter still touches it.
+struct UbrTimerTask {
+    UbrTimerId* slot;
+    std::atomic<bthread_timer_t> id;
+    void* (*cb)(void*);
+    void* arg;
+    UbrTimerBackoffFn backoff;
+    uint64_t interval_us;                        // timer thread only
+    bool periodic;
+    std::atomic<int> state;                      // kStarting/kScheduled/kDead
+    std::atomic<bool> stopped;
+    std::atomic<int> ref;
+    std::atomic<bool> join_pending;              // a DelAndWait is waiting
+    std::atomic<bool> done;                      // refs hit zero, joiner frees
+};
+
+namespace {
+
+void ReleaseRef(UbrTimerTask* task) {
+    if (task->ref.fetch_sub(1) == 1) {
+        if (task->join_pending.load()) {
+            task->done.store(true);              // joiner frees the task
+        } else {
+            delete task;
         }
-        return UBRING_ERR;
-    }
-    g_timer_module_initialized = 1;
-    return UBRING_OK;
-}
-
-void *UnifiedCallback(void *args) {
-    TimerFdCtx *ctx = (TimerFdCtx *)args;
-    if (pthread_spin_lock(&ctx->spin_lock) != 0) {
-        return nullptr;
-    }
-
-    if (ctx->status == TIMER_CONTEXT_NOT_USING) {
-        pthread_spin_unlock(&ctx->spin_lock);
-        return nullptr;
-    }
-
-    void *(*cb)(void *) = ctx->cb;
-    void *cb_args = ctx->args;
-    uint32_t fd = ctx->fd;
-    int is_periodical = ctx->periodical;
-    ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING;
-
-    pthread_spin_unlock(&ctx->spin_lock);
-
-    cb(cb_args);
-
-    if (!is_periodical) {
-        DeleteTimerInner(fd);
     }
-    return nullptr;
 }
 
-void *TimerEpoll(void *args) {
-    UNREFERENCE_PARAM(args);
-#if defined(OS_LINUX)
-    struct epoll_event ready_events[MAX_TIMER];
-#elif defined(OS_MACOSX)
-    struct kevent ready_events[MAX_TIMER];
-#endif
+void UbrTimerOnFire(void* p) {
+    UbrTimerTask* task = (UbrTimerTask*)p;
 
-    while (1) {
-        if (g_timer_module_initialized <= 0) {
-            LOG(ERROR) << "The Timer module is not initialized.";
-            break;
+    if (task->periodic) {
+        if (!task->stopped.load()) {
+            task->cb(task->arg);
         }
-
-#if defined(OS_LINUX)
-        int32_t ready_num = epoll_wait(g_epoll_fd, ready_events, MAX_TIMER,
-                                      TIMER_EPOLL_WAIT_TIMEOUT);
-#elif defined(OS_MACOSX)
-        struct timespec timeout = {0, TIMER_EPOLL_WAIT_TIMEOUT * 1000000};
-        int32_t ready_num = kevent(g_epoll_fd, nullptr, 0, ready_events, MAX_TIMER, &timeout);
-#endif
-
-        if (UNLIKELY(ready_num == -1)) {
-            errno_t err = errno;
-            if (err == EINTR) {
-                LOG_EVERY_SECOND(WARNING) << "Epoll/Kqueue wait was interrupted. errno=" << err;
-                continue;
-            } else if (err == EBADF) {
-                LOG(WARNING) << "The Timer module is destroyed.";
-                break;
+        // Claim the next schedule's ref before re-reading `stopped' so a
+        // racing delete can neither free the task nor orphan a re-arm.
+        task->ref.fetch_add(1);
+        if (task->stopped.load()) {
+            ReleaseRef(task);
+        } else {
+            uint64_t interval = task->interval_us;
+            if (task->backoff != nullptr) {
+                interval = task->backoff(task->arg, interval);
+                task->interval_us = interval;
             }
-            LOG(ERROR) << "Epoll/Kqueue wait internal error. errno=" << err;
-            break;
-        }
-
-        for (int32_t i = 0; i < ready_num; i++) {
-#if defined(OS_LINUX)
-            struct epoll_event *event = &ready_events[i];
-            int32_t timer_fd = event->data.fd;
-#elif defined(OS_MACOSX)
-            struct kevent *event = &ready_events[i];
-            int32_t timer_fd = event->ident;
-#endif
-
-            uint64_t exp = 0;
-            if (read(timer_fd, &exp, sizeof(exp)) < 0) {
-                if (errno != EBADF) {
-                    LOG(ERROR) << "Failed to read timerfd=" << timer_fd << " errno=" << errno;
+            bthread_timer_t id = 0;
+            if (bthread_timer_add(
+                    &id, butil::microseconds_from_now((int64_t)interval),
+                    UbrTimerOnFire, task) == 0) {
+                task->id.store(id);
+                if (task->stopped.load() && bthread_timer_del(id) == 0) {
+                    ReleaseRef(task);
                 }
-                continue;
-            }
-            if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) {
-                continue;
-            }
-
-            RETURN_CODE ret = ExecuteCallback(timer_fd);
-            if (ret != UBRING_OK) {
-                LOG(ERROR) << "Failed execute callback ret=" << ret;
-                DeleteTimerInner((uint32_t)timer_fd);
-                continue;
+            } else {
+                LOG(ERROR) << "Fail to re-arm ubring timer";
+                ReleaseRef(task);
             }
         }
-    }
-    return nullptr;
-}
-
-void DeleteTimerSafe(uint32_t fd) {
-    if (g_timer_fd_ctx_map == nullptr) {
+        ReleaseRef(task);
         return;
     }
 
-    if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) {
-        return;
+    // One-shot: exit the handle slot first -- after this the wrapper never
+    // touches the storage again, so the callback may release the object
+    // that holds it. Whether the callback runs is decided solely by this
+    // slot competition: every UbrTimerDel that wants the callback
+    // suppressed has to win this exchange first, so owned==true guarantees
+    // no UbrTimerDel is pending. Do not consult `stopped' here: its store
+    // (del thread) and this load (timer thread) are separated by the slot
+    // RMW and seq_cst does not order the store-buffer case -- ownership of
+    // the slot is the single arbiter.
+    UbrTimerId expected = task;
+    const bool owned =
+        __atomic_compare_exchange_n(task->slot, &expected, (UbrTimerId) nullptr,
+                                    false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
+    if (owned) {
+        task->cb(task->arg);
+    }
+    ReleaseRef(task);                            // schedule
+    if (owned) {
+        ReleaseRef(task);                        // owner
     }
-
-    if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) {
-        pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock);
-        return;
-    }
-
-    g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING;
-    g_timer_fd_ctx_map[fd].cb = nullptr;
-    g_timer_fd_ctx_map[fd].args = nullptr;
-    g_timer_fd_ctx_map[fd].periodical = 0;
-    g_timer_fd_ctx_map[fd].fd = 0;
-
-    pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock);
-
-#if defined(OS_LINUX)
-    epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr);
-#elif defined(OS_MACOSX)
-    struct kevent evt;
-    EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr);
-    kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr);
-#endif
-
-    uint64_t exp = 0;
-    read((int)fd, &exp, sizeof(exp));
-
-    close((int)fd);
-    std::atomic_fetch_sub(&g_total_timer_num, 1U);
 }
 
-void DeleteTimer(uint32_t fd) {
-    if (g_timer_fd_ctx_map == nullptr) {
-        LOG(WARNING) << "The timer is not initialized.";
-        return;
-    }
-
-    g_timer_fd_ctx_map[fd].periodical = 0;
+UbrTimerTask* TakeOutTask(UbrTimerId* slot) {
+    return __atomic_exchange_n(slot, (UbrTimerId) nullptr, __ATOMIC_SEQ_CST);
 }
 
-int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args) {
-    if (g_epoll_fd == -1) {
-        LOG(ERROR) << "Timer epoll/kqueue encountered internal error.";
-        return -1;
-    }
-
-#if defined(OS_LINUX)
-    int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
-#elif defined(OS_MACOSX)
-    int timer_fd = timerfd_create_macosx(CLOCK_MONOTONIC, 0);
-#endif
-
-    if (UNLIKELY(timer_fd >= (int)g_max_system_fd || timer_fd == -1)) {
-        LOG(ERROR) << "Failed to create timerfd=" << timer_fd << " errno=" << errno;
-        return -1;
+RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us,
+                               uint64_t interval_us, void* (*cb)(void*),
+                               void* arg, UbrTimerBackoffFn backoff) {
+    if (UNLIKELY(slot == nullptr || cb == nullptr)) {
+        LOG(ERROR) << "Ubr timer start invalid argument, slot=" << slot;
+        return UBRING_ERR;
     }
 
-    g_timer_fd_ctx_map[timer_fd].status = TIMER_CONTEXT_EPOLL_WAITING;
-    g_timer_fd_ctx_map[timer_fd].cb = cb;
-    g_timer_fd_ctx_map[timer_fd].args = args;
-    g_timer_fd_ctx_map[timer_fd].fd = (uint32_t)timer_fd;
-
-    if (LIKELY(time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0)) {
-        g_timer_fd_ctx_map[timer_fd].periodical = 1;
+    UbrTimerTask* task = new (std::nothrow) UbrTimerTask();
+    if (UNLIKELY(task == nullptr)) {
+        LOG(ERROR) << "Fail to malloc ubring timer task.";
+        return UBRING_ERR;
     }
-
-#if defined(OS_LINUX)
-    struct epoll_event event = {
-        .events = EPOLLIN,
-        .data = {.fd = timer_fd}
-    };
-
-    int32_t ret = epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, timer_fd, &event);
-#elif defined(OS_MACOSX)
-    struct kevent event;
-    uint64_t timeout_nsec = time->it_value.tv_sec * 1000000000ULL + time->it_value.tv_nsec;
-    uint64_t interval_nsec = time->it_interval.tv_sec * 1000000000ULL + time->it_interval.tv_nsec;
-    EV_SET(&event, timer_fd, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0,
-           timeout_nsec / 1000000, nullptr);
-    int32_t ret = kevent(g_epoll_fd, &event, 1, nullptr, 0, nullptr);
-#endif
-
-    if (UNLIKELY(ret != 0)) {
-        CloseTimerFd(timer_fd);
-        LOG(ERROR) << "Failed to add event to epoll/kqueue. errno=" << errno;
-        return -1;
+    task->slot = slot;
+    task->id.store(0);
+    task->cb = cb;
+    task->arg = arg;
+    task->backoff = backoff;
+    task->interval_us = interval_us;
+    task->periodic = (interval_us > 0);
+    task->state.store(kStarting);
+    task->stopped.store(false);
+    task->ref.store(3);                          // owner + schedule + starter
+    task->join_pending.store(false);
+    task->done.store(false);
+
+    // Publish the real task before scheduling so a delete or a DelAndWait
+    // racing the start always has an object to act on or wait for.
+    UbrTimerId expected = nullptr;
+    if (!__atomic_compare_exchange_n(slot, &expected, (UbrTimerId) task, false,
+                                     __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+        LOG(ERROR) << "Ubr timer start refused, slot already occupied";
+        delete task;                             // never published
+        return UBRING_ERR;
     }
 
-    std::atomic_fetch_add(&g_total_timer_num, 1U);
-
-#if defined(OS_LINUX)
-    ret = timerfd_settime(timer_fd, 0, time, nullptr);
-#elif defined(OS_MACOSX)
-    ret = timerfd_settime_macosx(timer_fd, 0, time, nullptr);
-#endif
-
-    if (UNLIKELY(ret != 0)) {
-#if defined(OS_LINUX)
-        if (epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, timer_fd, nullptr) != 0) {
-#elif defined(OS_MACOSX)
-        struct kevent evt;
-        EV_SET(&evt, timer_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr);
-        if (kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr) != 0) {
-#endif
-            LOG(ERROR) << "Failed to delete the timer fd=" << timer_fd << " with errno=" << errno;
+    bthread_timer_t id = 0;
+    if (UNLIKELY(bthread_timer_add(
+            &id, butil::microseconds_from_now((int64_t)delay_us),
+            UbrTimerOnFire, task) != 0)) {
+        LOG(ERROR) << "Fail to add ubring timer";
+        task->state.store(kDead);                // wake DelAndWait waiters
+        expected = task;
+        const bool owned =
+            __atomic_compare_exchange_n(slot, &expected, (UbrTimerId) nullptr,
+                                        false, __ATOMIC_SEQ_CST,
+                                        __ATOMIC_SEQ_CST);
+        ReleaseRef(task);                        // schedule, never ran
+        if (owned) {
+            ReleaseRef(task);                    // owner
         }
-        CloseTimerFd(timer_fd);
-        std::atomic_fetch_sub(&g_total_timer_num, 1U);
-        LOG(ERROR) << "Failed to set timer";
-        return -1;
+        ReleaseRef(task);                        // starter
+        return UBRING_ERR;
     }
-
-    return timer_fd;
+    // A zero-delay task may have fired and re-armed already; keep a newer
+    // id if so.
+    bthread_timer_t expected_id = 0;
+    task->id.compare_exchange_strong(expected_id, id);
+    task->state.store(kScheduled);
+    // No post-add stopped check here: a UbrTimerDel racing the start
+    // returns 1 without consuming the per-task resources, and the armed
+    // timer must fire so that OnFire settles the ownership protocol.
+    ReleaseRef(task);                            // starter
+    return UBRING_OK;
 }
 
-uint32_t GetActiveTimerNum(void) {
-    return std::atomic_load(&g_total_timer_num);
-}
+}  // namespace
 
-void CloseTimerFd(int fd) {
-    g_timer_fd_ctx_map[fd].cb = nullptr;
-    g_timer_fd_ctx_map[fd].args = nullptr;
-    g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING;
-    g_timer_fd_ctx_map[fd].fd = 0;
-    g_timer_fd_ctx_map[fd].periodical = 0;
-    if (close((int)fd) != 0) {
-        LOG(ERROR) << "Failed to close timer fd=" << fd << " errno=" << errno;
-        return;
-    }
+RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us,
+                          uint64_t interval_us, void* (*cb)(void*),
+                          void* arg, UbrTimerBackoffFn backoff) {
+    return TimerStartInternal(slot, delay_us, interval_us, cb, arg, backoff);
 }
 
-void TimerModuleDestroy(void) {
-    uint32_t max_fd = g_max_system_fd;
-    if (g_timer_fd_ctx_map) {
-        for (uint32_t fd = 0; fd < max_fd; fd++) {
-            if (g_timer_fd_ctx_map[fd].status != TIMER_CONTEXT_NOT_USING) {
-                DeleteTimerSafe(fd);
-            }
-        }
-    }
-    close(g_epoll_fd);
-    g_epoll_fd = -1;
-    g_total_timer_num = 0;
-    g_timer_module_initialized = 0;
-    int32_t ret = pthread_join(g_epoll_execute_thread, nullptr);
-    if (ret != EOK) {
-        LOG(ERROR) << "Failed to join pthread, during destroying timer module. ret=" << ret;
-        return;
-    }
+int UbrTimerDel(UbrTimerId* slot) {
+    if (slot == nullptr) {
+        return 1;
+    }
+    // Take the ownership of the slot first: after this exchange every
+    // dereference below is safe (the task cannot be freed while we hold
+    // the owner reference the slot used to anchor).
+    UbrTimerTask* task = TakeOutTask(slot);
+    if (task == nullptr) {
+        return 1;        // fired and cleared its slot (callback side consumed)
+                         // or another del won the exchange (it consumes)
+    }
+    task->stopped.store(true);                   // meaningful for periodic only
+    // A start still in flight cannot be cancelled nor dispatched yet; wait
+    // for the starter to settle the fate (kScheduled/kDead). Bounded: the
+    // starter stores the state before taking any lock our caller holds.
+    while (task->state.load() == kStarting) {
+        bthread_usleep(1000);
+    }
+    if (task->state.load() == kDead) {
+        ReleaseRef(task);                        // owner; schedule/starter are
+        return 1;                                // settled by the kDead path
+    }
+    bthread_timer_t id = task->id.load();
+    if (id != 0 && bthread_timer_del(id) == 0) {
+        ReleaseRef(task);                        // schedule: cancelled before dispatch
+    }                                            // ==1: dispatched, OnFire (owned==false)
+                                                 // releases it
+    ReleaseRef(task);                            // owner
+    return 0;       // This call won the slot competition. For a one-shot timer,
+                    // the callback will not run. For a periodic timer, future
+                    // rearming is stopped, but an already dispatched or running
+                    // callback may still complete.
 }
 
-RETURN_CODE TimerFdCtxValidate(uint32_t fd) {
-    if (fd >= g_max_system_fd) {
-        LOG(ERROR) << "TimerFd=" << fd << " is out of range=" << g_max_system_fd;
-        return UBRING_ERR;
-    }
-    if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) {
-        LOG(ERROR) << "TimerFd=" << fd << " has wrong status=" << g_timer_fd_ctx_map[fd].status;
-        return UBRING_ERR;
+void UbrTimerDelAndWait(UbrTimerId* slot) {
+    if (slot == nullptr) {
+        return;
     }
-    if (g_timer_fd_ctx_map[fd].cb == nullptr) {
-        LOG(ERROR) << "The callback is not set.";
-        return UBRING_ERR;
+    UbrTimerTask* task = TakeOutTask(slot);
+    if (task == nullptr) {
+        return;
     }
-
-    return UBRING_OK;
-}
-
-#if defined(OS_MACOSX)
-static int timerfd_create_macosx(int clockid, int flags) {
-    int pipefd[2];
-    if (pipe(pipefd) == -1) {
-        return -1;
+    task->join_pending.store(true);
+    task->stopped.store(true);
+    // A start still in flight cannot be cancelled yet; wait for the
+    // starter to schedule it or mark it dead.
+    while (task->state.load() == kStarting) {
+        bthread_usleep(1000);
+    }
+    if (task->state.load() == kScheduled) {
+        bthread_timer_t id = task->id.load();
+        if (id != 0 && bthread_timer_del(id) == 0) {
+            ReleaseRef(task);                    // cancelled before run
+        }
     }
-    return pipefd[0];
-}
-
-static int timerfd_settime_macosx(int fd, int flags,
-                                   const itimerspec *new_value,
-                                   itimerspec *old_value) {
-    if (old_value != nullptr) {
-        memset(old_value, 0, sizeof(itimerspec));
+    ReleaseRef(task);                            // owner reference
+    while (!task->done.load()) {
+        bthread_usleep(1000);
     }
-    return 0;
+    task->join_pending.store(false);
+    delete task;
 }
-#endif
 
 }  // namespace ubring
-}  // namespace brpc
\ No newline at end of file
+}  // namespace brpc
diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h
index 1b42caef04..f4d244eceb 100644
--- a/src/brpc/ubshm/timer/timer_mgr.h
+++ b/src/brpc/ubshm/timer/timer_mgr.h
@@ -15,59 +15,62 @@
 // specific language governing permissions and limitations
 // under the License.
 
+// bthread based timer facade for the ubring module. Callbacks run on the
+// process-wide bthread timer thread and must return quickly.
+
 #ifndef BRPC_TIMER_MGR_H
 #define BRPC_TIMER_MGR_H
-#include <pthread.h>
-#include <time.h>
-#include "brpc/ubshm/common/common.h"
-
-#if defined(OS_LINUX)
-#include <sys/epoll.h>
-#include <sys/timerfd.h>
-#elif defined(OS_MACOSX)
-#include <sys/types.h>
-#include <sys/event.h>
-#include <sys/time.h>
-#endif
 
-#define MAX_TIMER 1024
-#define TIMER_EPOLL_WAIT_TIMEOUT 1000
+#include <stdint.h>
+#include "brpc/ubshm/common/common.h"
 
-#if defined(OS_MACOSX)
-struct itimerspec
-{
-    struct timespec it_interval;
-    struct timespec it_value;
-};
-#endif
 namespace brpc {
 namespace ubring {
-typedef enum {
-    TIMER_CONTEXT_NOT_USING,
-    TIMER_CONTEXT_EPOLL_WAITING,
-    TIMER_CONTEXT_CALLBACK_ONGOING
-} TimerFdCtxStatus;
 
-typedef struct {
-    void *(*cb)(void*);
-    void *args;
-    uint32_t fd;
-    TimerFdCtxStatus status;
-    uint32_t periodical;
-    pthread_spinlock_t spin_lock;
-} TimerFdCtx;
+// Opaque timer handle. nullptr means "not started" (or already deleted /
+// fired for one-shot timers).
+typedef struct UbrTimerTask* UbrTimerId;
+
+// Maps the current re-arm interval of a periodic timer to the next one.
+// Runs on the timer thread only.
+typedef uint64_t (*UbrTimerBackoffFn)(void* arg, uint64_t cur_interval_us);
+
+// Schedule `cb(arg)' to run after `delay_us' and, when `interval_us' > 0,
+// re-arm itself after every run until deleted. One-shot timers release
+// their handle slot before running the callback, so the callback may free
+// the object that stores the slot; the task object itself is released
+// automatically.
+RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us,
+                          uint64_t interval_us, void* (*cb)(void*),
+                          void* arg,
+                          UbrTimerBackoffFn backoff = nullptr);
+
+// Non-blocking delete, safe to call from inside the timer callback itself.
+// This function does not wait for an already running callback and does not
+// protect resources reachable from `arg` on its own.
+//
+// Returns 0 when this call wins the handle-slot competition.
+// - For a one-shot timer, the callback will not run.
+// - For a periodic timer, future rearming is stopped, but an already
+//   dispatched or running callback may still execute once more. Callers
+//   must not reclaim resources reachable from `arg` based on this return
+//   alone; use UbrTimerDelAndWait when teardown needs to wait for callbacks.
+//
+// Returns 1 when this caller did not acquire timer ownership. The callback,
+// another deleter, or the scheduling path is responsible for settling the
+// timer resources, so this caller must not reclaim them.
+int UbrTimerDel(UbrTimerId* slot);
+
+// Delete and wait until a possibly running callback finished, so the
+// caller can free resources reachable from `arg'. Never call this on the
+// callback's own task. A one-shot callback that is already running holds
+// the ownership of `arg' by itself (mirroring bthread_timer_del returning
+// 1) and cannot be waited for through the slot. The wait polls with
+// bthread_usleep, which degrades to ::usleep on plain pthread callers
+// (e.g. process-exit paths).
+void UbrTimerDelAndWait(UbrTimerId* slot);
 
-RETURN_CODE TimerInit(void);
-void TimerModuleDestroy(void);
-void *UnifiedCallback(void *args);
-void *TimerEpoll(void *args);
-int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args);
-uint32_t GetActiveTimerNum(void);
-void CloseTimerFd(int fd);
+}  // namespace ubring
+}  // namespace brpc
 
-void DeleteTimerSafe(uint32_t fd);
-void DeleteTimer(uint32_t fd);
-RETURN_CODE TimerFdCtxValidate(uint32_t fd);
-}
-}
-#endif //BRPC_TIMER_MGR_H
\ No newline at end of file
+#endif //BRPC_TIMER_MGR_H
diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp
index 31539fda85..089ccef5bd 100644
--- a/src/brpc/ubshm/ub_endpoint.cpp
+++ b/src/brpc/ubshm/ub_endpoint.cpp
@@ -848,6 +848,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag,
         while (running->load(std::memory_order_relaxed)) {
             while (poller->op_queue.Dequeue(op)) {
                 if (op.type == PollerSidOp::ADD) {
+                    poller_sids.erase(op);
                     poller_sids.emplace(op);
                 } else if (op.type == PollerSidOp::REMOVE) {
                     poller_sids.erase(op);
diff --git a/src/brpc/ubshm/ub_helper.cpp b/src/brpc/ubshm/ub_helper.cpp
index 230ca71bac..d3c769140d 100644
--- a/src/brpc/ubshm/ub_helper.cpp
+++ b/src/brpc/ubshm/ub_helper.cpp
@@ -59,10 +59,7 @@ static void GlobalUBInitializeOrDieImpl() {
         ExitWithError();
     }
 
-    if (TimerInit()) {
-        PLOG(ERROR) << "Fail to TimerInit";
-        ExitWithError();
-    }
+    // Timers are bthread based and start lazily on first use.
 
     if (ShmMgrInit()) {
         PLOG(ERROR) << "Fail to ShmMgrInit";
diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp
index ea34267332..25b6f2749b 100644
--- a/src/brpc/ubshm/ub_ring.cpp
+++ b/src/brpc/ubshm/ub_ring.cpp
@@ -19,6 +19,7 @@
 #include <gflags/gflags.h>
 #include <unistd.h>
 #include <ctime>
+#include <new>
 #include "bthread/bthread.h"
 #include "butil/logging.h"
 #include "brpc/ubshm/ub_ring.h"
@@ -38,6 +39,14 @@ DEFINE_int32(ub_hb_retry_cnt, 10,
              "UBRing heartbeat retry count.");
 DEFINE_int32(ub_event_queue_timer_interval_us, 100,
              "UBRing disconnection check interval in microseconds.");
+DEFINE_int32(ub_event_queue_timer_interval_max_us, 10000,
+             "UBRing upper bound of the close-check polling interval in "
+             "microseconds while the link is idle; the interval backs off "
+             "from ub_event_queue_timer_interval_us up to this value. "
+             "Set to 0 to keep the interval steady (back-off disabled).");
+
+// Exponential back-off multiplier of the close-check polling interval.
+constexpr uint64_t kCloseCheckBackoffFactor = 2;
 
 UBRing::UBRing()
 {}
@@ -59,6 +68,122 @@ RETURN_CODE UBRing::UbrTrxMapShm(SHM *local_shm, SHM *remote_shm)
     return UBRING_OK;
 }
 
+static void UbrDoAsynClearWork(UbrTrx *trx, uint64_t expect_ubr_id) {
+    if (UNLIKELY(UBRing::UbrTrxFreeShm(trx) != UBRING_OK)) {
+        LOG(ERROR) << "Trx close, wait for local shm " << trx->local_shm.name << " free fail.";
+    }
+    if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) {
+        LOG(ERROR) << "Trx close, release shm " << trx->local_shm.name << " trx failed.";
+    }
+}
+
+static void UbrDoPassiveClearWork(UbrTrx *trx, uint64_t expect_ubr_id) {
+    int rc = ShmLocalFree(&trx->remote_shm);
+    if (rc != UBRING_OK) {
+        LOG(ERROR) << "Trx passive clear, delete remote shm " << trx->remote_shm.name
+                   << " failed. ret=" << rc;
+    }
+    rc = ShmLocalFree(&trx->local_shm);
+    if (rc != UBRING_OK) {
+        LOG(ERROR) << "Trx passive clear, delete local shm " << trx->local_shm.name
+                   << " failed. ret=" << rc;
+    }
+    if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) {
+        LOG(ERROR) << "Trx passive clear, release shm " << trx->local_shm.name << " trx failed.";
+    }
+}
+
+// Schedule the delayed cleanup of `trx'. The cleanup ownership lives in the
+// per-acquisition control object, so exactly one of the delayed-clear
+// callback and a force close ever runs the cleanup. `work' is the cleanup
+// body, used directly when the timer cannot be started.
+static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*),
+                                         void (*work)(UbrTrx*, uint64_t)) {
+    if (UNLIKELY(trx == nullptr || trx->local_shm.addr == nullptr)) {
+        return UBRING_OK;                    // released trx, stale event
+    }
+    if (__atomic_load_n(&trx->cleanup_ctl, __ATOMIC_SEQ_CST) != nullptr) {
+        return UBRING_OK;                    // cleanup already scheduled
+    }
+    auto* ctl = new (std::nothrow) UbrCleanupCtl();
+    if (UNLIKELY(ctl == nullptr)) {
+        LOG(ERROR) << "Fail to malloc ubr cleanup ctl.";
+        return UBRING_ERR;
+    }
+    ctl->trx = trx;
+    ctl->ubr_id = ATOMIC_LOAD(trx->ubr_id);
+    ctl->state.store(UBR_CLEANUP_PENDING);
+    ctl->timer = nullptr;
+    ctl->ref.store(2);                       // timer/callback + starter; the
+                                             // manager anchor is taken by
+                                             // TryPublishUnitCleanupCtl
+
+    UbrCleanupCtl* expected = nullptr;
+    if (!__atomic_compare_exchange_n(&trx->cleanup_ctl, &expected, ctl, false,
+                                     __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+        delete ctl;                          // another schedule won
+        return UBRING_OK;
+    }
+    if (!UBRingManager::TryPublishUnitCleanupCtl(trx->trx_mgr_index,
+                                                 ctl->ubr_id, ctl)) {
+        // The slot was released (and possibly reused) before we could
+        // anchor: force close or the new occupant owns it now. Nothing
+        // was armed yet -- just undo the trx-side publication.
+        expected = ctl;
+        __atomic_compare_exchange_n(&trx->cleanup_ctl, &expected,
+                                    (UbrCleanupCtl*) nullptr, false,
+                                    __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
+        ctl->ReleaseRef();                   // timer/callback reference, never armed
+        ctl->ReleaseRef();                   // starter reference
+        return UBRING_OK;
+    }
+    RETURN_CODE rc = UbrTimerStart(&ctl->timer,
+            (uint64_t)FLAGS_ub_flying_io_timeout_s * SEC_TO_USEC, 0, cb, ctl);
+    if (UNLIKELY(rc != UBRING_OK)) {
+        // The timer was never scheduled: this path owns the manager,
+        // timer/callback and starter references. Roll the schedule back
+        // and run the cleanup inline so the trx does not end up with
+        // neither timers nor a queued cleanup. If force close claimed the
+        // ownership meanwhile, leave the manager anchor to it.
+        int state_expected = UBR_CLEANUP_PENDING;
+        if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, state_expected, UBR_CLEANUP_RUNNING)) {
+            if (ATOMIC_LOAD(trx->ubr_id) == ctl->ubr_id) {
+                work(trx, ctl->ubr_id);
+            }
+            ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+            // Detach only after DONE: while the inline cleanup runs, the
+            // anchor and cleanup_ctl must keep telling a concurrent force
+            // close that this cleanup is owned (RUNNING/DONE), otherwise it
+            // would fall into its no-ctl branch and clean the trx again.
+            UbrCleanupCtl* published = ctl;
+            __atomic_compare_exchange_n(&trx->cleanup_ctl, &published,
+                                        (UbrCleanupCtl*) nullptr, false,
+                                        __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
+            UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl);
+        }
+        ctl->ReleaseRef();                   // timer/callback reference
+        ctl->ReleaseRef();                   // starter reference
+        return UBRING_ERR;
+    }
+    if (UNLIKELY(ATOMIC_LOAD(trx->ubr_id) != ctl->ubr_id)) {
+        // Published onto a slot that was released and reused meanwhile.
+        if (UbrTimerDel(&ctl->timer) == 0) {
+            ctl->ReleaseRef();               // timer/callback reference
+        }
+        UbrCleanupCtl* published = ctl;
+        if (!__atomic_compare_exchange_n(&trx->cleanup_ctl, &published,
+                                         (UbrCleanupCtl*) nullptr, false,
+                                         __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+            published = nullptr;
+        }
+        UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl);
+        ctl->ReleaseRef();                   // starter reference
+        return UBRING_OK;
+    }
+    ctl->ReleaseRef();                       // starter reference
+    return UBRING_OK;
+}
+
 RETURN_CODE UBRing::UbrTrxClose() {
     RETURN_CODE close_check_rc = UbrTrxCloseCheck(_trx);
     if (UNLIKELY(close_check_rc != UBRING_OK)) {
@@ -89,17 +214,46 @@ RETURN_CODE UBRing::UbrTrxClose() {
             LOG(WARNING) << "Local shm " << _trx->local_shm.name
             << " wait for the peer to close timed out, force cleanup.";
             _trx->ubr_rx.trx_state = UBR_STATE_CLOSED;
-            // Force synchronous cleanup instead of relying on async timer
-            DeleteTimerSafe((uint32_t)_trx->timer_fd);
-            DeleteTimerSafe((uint32_t)_trx->hb_timer_fd);
-            if (_trx->ubr_tx.remote_rx_event_q.addr != nullptr) {
-                ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->flag = UBR_STATE_CLOSED;
+            // Wait out the close/heartbeat callbacks, which may schedule a
+            // delayed cleanup, then settle the cleanup ownership: force
+            // runs the cleanup itself when it can claim it, and leaves it
+            // to an already running delayed-clear callback otherwise.
+            const uint64_t expect_ubr_id = ATOMIC_LOAD(_trx->ubr_id);
+            UbrTimerDelAndWait(&_trx->close_timer);
+            UbrTimerDelAndWait(&_trx->hb_timer);
+            UbrCleanupCtl* ctl = UBRingManager::SnapshotUnitCleanupCtl(_trx->trx_mgr_index);
+            if (ctl != nullptr && ctl->ubr_id != expect_ubr_id) {
+                ctl->ReleaseRef();               // snapshot reference
+                ctl = nullptr;                   // slot reused, not ours
+            }
+            bool cleanup_owned = false;
+            if (ctl != nullptr) {
+                int expected = UBR_CLEANUP_PENDING;
+                if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, expected, UBR_CLEANUP_RUNNING)) {
+                    cleanup_owned = true;
+                    if (UbrTimerDel(&ctl->timer) == 0) {
+                        ctl->ReleaseRef();   // timer/callback reference
+                    }
+                }
+            } else if (ATOMIC_LOAD(_trx->ubr_id) == expect_ubr_id) {
+                cleanup_owned = true;
             }
-            if (UNLIKELY(UbrTrxFreeShm(_trx) != UBRING_OK)) {
-                LOG(WARNING) << "Force close, local shm " << _trx->local_shm.name << " free failed.";
+            if (cleanup_owned) {
+                if (_trx->ubr_tx.remote_rx_event_q.addr != nullptr) {
+                    ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->flag = UBR_STATE_CLOSED;
+                }
+                if (UNLIKELY(UbrTrxFreeShm(_trx) != UBRING_OK)) {
+                    LOG(WARNING) << "Force close, local shm " << _trx->local_shm.name << " free failed.";
+                }
+                if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(_trx, expect_ubr_id) != UBRING_OK)) {
+                    LOG(WARNING) << "Force close, release trx " << _trx->local_shm.name << " failed.";
+                }
+                if (ctl != nullptr) {
+                    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+                }
             }
-            if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(_trx) != UBRING_OK)) {
-                LOG(WARNING) << "Force close, release trx " << _trx->local_shm.name << " failed.";
+            if (ctl != nullptr) {
+                ctl->ReleaseRef();               // snapshot reference
             }
             return UBRING_ERR_TIMEOUT;
         }
@@ -107,7 +261,7 @@ RETURN_CODE UBRing::UbrTrxClose() {
     }
     _trx->ubr_rx.trx_state = UBR_STATE_CLOSED;
     RETURN_CODE rc;
-    if (UNLIKELY((rc = ClearTrxResource(_trx, start_time, UBR_SEND_CLOSE)) != UBRING_OK)) {
+    if (UNLIKELY((rc = ClearTrxResource(_trx)) != UBRING_OK)) {
         if (rc == UBRING_REENTRY) {
             LOG(INFO) << "Trx close, peer is closing, trx local name=" << _trx->local_shm.name;
             return UBRING_OK;
@@ -124,24 +278,51 @@ RETURN_CODE UBRing::UbrTrxClose() {
     return UBRING_OK;
 }
 
+// Back-off policy of the close-check timer: fast while there is traffic or
+// a close in progress, doubling up to the cap while idle.
+static uint64_t UbrCloseTimerBackoff(void* arg, uint64_t cur_interval_us) {
+    auto* trx = (UbrTrx*)arg;
+    auto* local_rx_event_q = (UbrEventQMsg *)trx->ubr_rx.local_rx_event_q.addr;
+    auto* local_tx_event_q = (UbrEventQMsg *)trx->ubr_tx.local_tx_event_q.addr;
+    const uint64_t in_io_id = ATOMIC_LOAD(trx->ubr_rx.in_io_id);
+    const uint64_t out_io_id = ATOMIC_LOAD(trx->ubr_tx.out_io_id);
+    if (UNLIKELY(local_rx_event_q == nullptr)) {
+        return (uint64_t)FLAGS_ub_event_queue_timer_interval_us;
+    }
+    const bool has_traffic = (in_io_id != trx->close_chk_in_io_id) ||
+                             (out_io_id != trx->close_chk_out_io_id);
+    const bool closing = (local_rx_event_q->flag != UBR_STATE_CONNECTED);
+    trx->close_chk_in_io_id = in_io_id;
+    trx->close_chk_out_io_id = out_io_id;
+    if (has_traffic || closing || local_tx_event_q == nullptr) {
+        return (uint64_t)FLAGS_ub_event_queue_timer_interval_us;
+    }
+    uint64_t next = cur_interval_us * kCloseCheckBackoffFactor;
+    const uint64_t max_us = (uint64_t)FLAGS_ub_event_queue_timer_interval_max_us;
+    if (max_us > 0 && next > max_us) {
+        next = max_us;
+    } else if (max_us == 0) {
+        next = cur_interval_us;
+    }
+    return next;
+}
+
 RETURN_CODE UBRing::UbrAddCloseTimer() {
     if (UNLIKELY(_trx == nullptr)) {
         LOG(ERROR) << "Trx add close timer failed, trx is null.";
         return UBRING_ERR;
     }
 
-    const uint32_t event_q_timer_interval_ns =
-        FLAGS_ub_event_queue_timer_interval_us * USEC_TO_NSEC;
-    itimerspec time_spec = {
-            .it_interval = {.tv_sec = 0, .tv_nsec = event_q_timer_interval_ns},
-            .it_value = {.tv_sec = 0, .tv_nsec = 1}
-    };
-    int timer_fd = TimerStart(&time_spec, UbrTrxCloseCallback, (void*)_trx);
-    if (UNLIKELY(timer_fd == -1)) {
+    const uint32_t interval_us = FLAGS_ub_event_queue_timer_interval_us;
+    _trx->close_chk_in_io_id = ATOMIC_LOAD(_trx->ubr_rx.in_io_id);
+    _trx->close_chk_out_io_id = ATOMIC_LOAD(_trx->ubr_tx.out_io_id);
+    RETURN_CODE rc = UbrTimerStart(&_trx->close_timer, 0, interval_us,
+                                   UbrTrxCloseCallback, (void*)_trx,
+                                   UbrCloseTimerBackoff);
+    if (UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start ubr close timer failed, trx local name=" << _trx->local_shm.name;
         return UBRING_ERR;
     }
-    _trx->timer_fd = timer_fd;
     return UBRING_OK;
 }
 
@@ -152,7 +333,7 @@ RETURN_CODE UBRing::UbrAddTimer() {
     }
 
     if (UNLIKELY(UbrAddHBTimer() != UBRING_OK)) {
-        DeleteTimerSafe((uint32_t)_trx->timer_fd);
+        UbrTimerDelAndWait(&_trx->close_timer);
         LOG(ERROR) << "Ubr " << _trx->local_shm.name << " add heartbeat timer failed.";
         return UBRING_ERR;
     }
@@ -171,15 +352,12 @@ void* UBRing::UbrTrxCloseCallback(void* args) {
         return nullptr;
     }
     trx->ubr_rx.trx_state = UBR_STATE_CLOSED;
-    int fd = (int)trx->local_shm.fd;
     do {
         if (ATOMIC_LOAD(trx->close_cnt) == 0) {
             break;
         }
         ATOMIC_SUB(trx->close_cnt, 1);
 
-        uint64_t start_time = GetCurNanoSeconds();
-
         if (local_tx_event_q->flag == UBR_STATE_CONNECTED || ATOMIC_LOAD(trx->close_cnt) == 1) {
             local_tx_event_q->flag = UBR_STATE_CLOSED;
             trx->ubr_tx.trx_state = UBR_STATE_CLOSED;
@@ -190,7 +368,7 @@ void* UBRing::UbrTrxCloseCallback(void* args) {
             break;
         }
         remote_rx_event_q->flag = UBR_STATE_CLOSED;
-        RETURN_CODE clear_rc = ClearTrxResource(trx, start_time, UBR_CALL_BACK_CLOSE, 1);
+        RETURN_CODE clear_rc = ClearTrxResource(trx);
         if (UNLIKELY(clear_rc != UBRING_OK && clear_rc != UBRING_REENTRY)) {
             LOG(ERROR) << "Trx close callback failed, " << trx->local_shm.name << " clear trx resource failed.";
             break;
@@ -205,54 +383,56 @@ RETURN_CODE UBRing::UbrAddHBTimer() {
         return UBRING_ERR;
     }
 
-    itimerspec time_spec = {
-            .it_interval = {.tv_sec = FLAGS_ub_hb_timer_interval_s, .tv_nsec = 0},
-            .it_value = {.tv_sec = 0, .tv_nsec = 1}
-    };
-    int timer_fd = TimerStart(&time_spec, UbrTrxHBCallback, (void*)_trx);
-    if (UNLIKELY(timer_fd == -1)) {
+    const uint64_t interval_us = (uint64_t)FLAGS_ub_hb_timer_interval_s * SEC_TO_USEC;
+    RETURN_CODE rc = UbrTimerStart(&_trx->hb_timer, 0, interval_us,
+                                   UbrTrxHBCallback, (void*)_trx);
+    if (UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start ubr heartbeat timer failed.";
         return UBRING_ERR;
     }
-    _trx->hb_timer_fd = timer_fd;
     return UBRING_OK;
 }
 
-RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx, int fd, PASSIVE_DISC_TYPE type) {
+RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx) {
     RETURN_CODE passive_close_check_rc = UbrTrxCloseCheck(trx);
     if (UNLIKELY(passive_close_check_rc != UBRING_OK)) {
         if (passive_close_check_rc == UBRING_REENTRY) {
             LOG(INFO) << "Passive close skipped, active close in progress, name=" << trx->local_shm.name;
-            uint64_t start_time = GetCurNanoSeconds();
-            return ClearTrxResource(trx, start_time, UBR_CALL_BACK_CLOSE);
+            return ClearTrxResource(trx);
         }
         return UBRING_ERR;
     }
     trx->ubr_tx.trx_state = UBR_STATE_CLOSED;
     trx->ubr_rx.trx_state = UBR_STATE_CLOSED;
-    DeleteTimerSafe((uint32_t)trx->timer_fd);
-    const char *type_name = nullptr;
-    if (type == UBR_HEARTBEAT) {
-        DeleteTimer((uint32_t)trx->hb_timer_fd);
-        type_name = "Trx heartbeat";
-    } else if (type == UBR_UB_EVENT) {
-        DeleteTimerSafe((uint32_t)trx->hb_timer_fd);
-        type_name = "Ub event callback";
-    }
-    constexpr int64_t kMicrosecondsPerSecond = 1000000LL;
-    bthread_usleep(FLAGS_ub_flying_io_timeout_s * kMicrosecondsPerSecond);
+    // Non-blocking: this may run inside the heartbeat callback itself.
+    UbrTimerDel(&trx->close_timer);
+    UbrTimerDel(&trx->hb_timer);
+    // Wait for in-flight IO on a one-shot timer instead of sleeping on the
+    // timer thread.
+    return UbrScheduleClearTimer(trx, UbrPassiveClearCallback, UbrDoPassiveClearWork);
+}
 
-    int rc = ShmLocalFree(&trx->remote_shm);
-    if (rc != UBRING_OK) {
-        LOG(ERROR) << type_name << ", delete remote shm failed. ret=" << rc;
+void* UBRing::UbrPassiveClearCallback(void* args) {
+    auto* ctl = (UbrCleanupCtl*)args;
+    if (UNLIKELY(ctl == nullptr)) {
+        LOG(ERROR) << "Trx passive clear callback failed, ctl is null.";
+        return nullptr;
     }
-    rc = ShmLocalFree(&trx->local_shm);
-    if (rc != UBRING_OK) {
-        LOG(ERROR) << type_name << ", delete local shm failed. ret=" << rc;
+    int expected = UBR_CLEANUP_PENDING;
+    if (!ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state,
+                                        expected, UBR_CLEANUP_RUNNING)) {
+        // Force close owns the cleanup; this fire still holds the
+        // timer/callback reference inherited from the schedule.
+        ctl->ReleaseRef();
+        return nullptr;
     }
-
-    UBRingManager::ReleaseUbrTrxFromMgr(trx);
-    return UBRING_OK;
+    UbrTrx* trx = ctl->trx;
+    if (UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) {
+        UbrDoPassiveClearWork(trx, ctl->ubr_id);
+    }
+    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+    ctl->ReleaseRef();                       // timer/callback reference
+    return nullptr;
 }
 
 void* UBRing::UbrTrxHBCallback(void* args) {
@@ -286,9 +466,9 @@ void* UBRing::UbrTrxHBCallback(void* args) {
     }
 
     int fd = (int)trx->local_shm.fd;
-    LOG(INFO) << "Hlc heartbeat, start to clear trx resource. hb_timer_fd=" << fd << ", shm_name=" << trx->local_shm.name;
-    UbrPassiveClearTrx(trx, fd, UBR_HEARTBEAT);
-    LOG(INFO) << "Hlc heartbeat clear trx resource finish.";
+    LOG(INFO) << "Ubr heartbeat, start to clear trx resource. shm_fd=" << fd << ", shm_name=" << trx->local_shm.name;
+    UbrPassiveClearTrx(trx);
+    LOG(INFO) << "Ubr heartbeat clear trx resource finish.";
     return nullptr;
 }
 
@@ -297,40 +477,30 @@ RETURN_CODE UBRing::UbrAddAsynClearTimer(UbrTrx *trx) {
         LOG(ERROR) << "Trx add close timer failed, trx is null.";
         return UBRING_ERR;
     }
-
-    if (trx->clear_timer_fd > 0) {
-        return UBRING_OK;
-    }
-
-    itimerspec time_spec = {
-            .it_interval = {.tv_sec = 0, .tv_nsec = 0},
-            .it_value = {.tv_sec = FLAGS_ub_flying_io_timeout_s, .tv_nsec = 0}
-    };
-
-    int timer_fd = TimerStart(&time_spec, UbrAsynClearCallback, (void*)trx);
-    if (UNLIKELY(timer_fd == -1)) {
-        LOG(ERROR) << "Start ubr close timer failed, trx name=%s.", trx->local_shm.name;
-        return UBRING_ERR;
-    }
-    trx->clear_timer_fd = timer_fd;
-    return UBRING_OK;
+    return UbrScheduleClearTimer(trx, UbrAsynClearCallback, UbrDoAsynClearWork);
 }
 
 void *UBRing::UbrAsynClearCallback(void *args)
 {
-    auto* trx = (UbrTrx*) args;
-    if (UNLIKELY(trx == nullptr)) {
-        LOG(ERROR) << "Trx close, trx is null.";
+    auto* ctl = (UbrCleanupCtl*) args;
+    if (UNLIKELY(ctl == nullptr)) {
+        LOG(ERROR) << "Trx close, ctl is null.";
         return nullptr;
     }
-
-    if (UNLIKELY(UbrTrxFreeShm(trx) != UBRING_OK)) {
-        LOG(ERROR) << "Trx close, wait for local shm " << trx->local_shm.name << " free fail.";
+    int expected = UBR_CLEANUP_PENDING;
+    if (!ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state,
+                                        expected, UBR_CLEANUP_RUNNING)) {
+        // Force close owns the cleanup; this fire still holds the
+        // timer/callback reference inherited from the schedule.
+        ctl->ReleaseRef();
+        return nullptr;
     }
-
-    if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx) != UBRING_OK)) {
-        LOG(ERROR) << "Trx close, release shm " << trx->local_shm.name << " trx failed.";
+    UbrTrx* trx = ctl->trx;
+    if (UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) {
+        UbrDoAsynClearWork(trx, ctl->ubr_id);
     }
+    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+    ctl->ReleaseRef();                       // timer/callback reference
     return nullptr;
 }
 
@@ -360,8 +530,8 @@ int UBRing::UbrTrxSend(const void *buf, uint32_t buf_len)
     uint32_t total_send_len = 0;
     uint32_t remain_buf_len = buf_len;
     uint8_t is_last_pkt = 0;
-    _trx->ubr_tx.out_io_id++;
-    ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->io_id = _trx->ubr_tx.out_io_id;
+    const uint64_t io_seq = ATOMIC_ADD(_trx->ubr_tx.out_io_id, 1) + 1;
+    ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->io_id = io_seq;
     while (remain_buf_len > 0) {
         is_last_pkt = (uint8_t)(remain_buf_len <= UBR_MSG_PAYLOAD_LEN);
         msg->header[UBR_MSG_FLAG_INDEX] = is_last_pkt ? UBR_MSG_CHUNK_EOF : UBR_MSG_CHUNK_EXIST;
@@ -474,8 +644,8 @@ ssize_t UBRing::UbrTrxWritev(const struct iovec *iov, int iovcnt)
     size_t fulled = 0;
     uint8_t is_last_pkt = 0;
     uint8_t cur_pkt_len = 0;
-    _trx->ubr_tx.out_io_id++;
-    ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->io_id = _trx->ubr_tx.out_io_id;
+    const uint64_t io_seq = ATOMIC_ADD(_trx->ubr_tx.out_io_id, 1) + 1;
+    ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->io_id = io_seq;
     while (buf_len > 0) {
         is_last_pkt = (uint8_t)(buf_len <= UBR_MSG_PAYLOAD_LEN);
         cur_pkt_len = is_last_pkt ? (uint8_t)buf_len : UBR_MSG_PAYLOAD_LEN;
@@ -563,7 +733,7 @@ RETURN_CODE UBRing::IsUbrTrxReadable(uint32_t ep_event)
     }
 
     uint64_t io_id = ((UbrEventQMsg *)_trx->ubr_rx.local_rx_event_q.addr)->io_id;
-    if ((ep_event & EPOLLET) && io_id == _trx->ubr_rx.in_io_id) {
+    if ((ep_event & EPOLLET) && io_id == ATOMIC_LOAD(_trx->ubr_rx.in_io_id)) {
         return MPA_MUXER_NOT_READY;
     }
 
@@ -578,7 +748,7 @@ RETURN_CODE UBRing::IsUbrTrxReadable(uint32_t ep_event)
         return MPA_MUXER_NOT_READY;
     }
     if (ep_event & EPOLLET) {
-        _trx->ubr_rx.in_io_id = io_id;
+        ATOMIC_STORE(_trx->ubr_rx.in_io_id, io_id);
     }
     return UBRING_OK;
 }
@@ -834,7 +1004,7 @@ int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm) {
     if (UNLIKELY((UbrServerTrxInit(local_trx_shm, remote_trx_shm)) != UBRING_OK)) {
         LOG(ERROR) << "Server trx init failed.";
         UbrTrxFreeShm(_trx);
-        UBRingManager::ReleaseUbrTrxFromMgr(_trx);
+        UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id));
         _trx = nullptr;
         return -1;
     }
@@ -887,7 +1057,8 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char *loca
         local_name,
         SERVER_SHM_NAME_SUFFIX);
     if (UNLIKELY(result < 0)) {
-        LOG(ERROR) << "Copy server shared memory name failed, local_name=%s, ret=%d.", local_name, result;
+        LOG(ERROR) << "Copy server shared memory name failed, local_name=" << local_name
+                   << ", ret=" << result;
         return UBRING_ERR;
     }
     UbrSetSleepTask(UBR_TASK_CONNECT_MAP_FRONT);
@@ -908,8 +1079,8 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char *loca
     uint32_t timeout = ((UbrDataStatusQMsg *)(_trx->ubr_tx.local_data_status_q.addr))->timeout;
     if (HasTimedOut(start_time, timeout) != UBRING_OK) {
         LOG(ERROR) << "Local shm " << local_trx_shm->name << " wait for connect remote map timeout.";
-        DeleteTimerSafe((uint32_t)_trx->hb_timer_fd);
-        DeleteTimerSafe((uint32_t)_trx->timer_fd);
+        UbrTimerDelAndWait(&_trx->hb_timer);
+        UbrTimerDelAndWait(&_trx->close_timer);
         ShmRemoteFree(&_trx->remote_shm);
         return UBRING_ERR_TIMEOUT;
     }
@@ -940,14 +1111,14 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_na
         if (rc == SHM_ERR_EXIST || rc == SHM_ERR_NOT_FOUND) {
             rc = UBR_ERR_ADDR_IN_USE;
         }
-        UBRingManager::ReleaseUbrTrxFromMgr(_trx);
+        UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id));
         return rc;
     }
     rc = UbrTrxMapLocalShm(local_trx_shm);
     if (UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Trx map local shared memory failed, local shm name=" << local_trx_shm->name;
         ShmLocalFree(local_trx_shm);
-        UBRingManager::ReleaseUbrTrxFromMgr(_trx);
+        UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id));
         return rc;
     }
     ((UbrDataStatusQMsg *)_trx->ubr_tx.local_data_status_q.addr)->timeout =
@@ -957,7 +1128,7 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_na
     if (rc != UBRING_OK) {
         LOG(ERROR) << "Get ubring deal msg max cnt, local shm name=" << local_trx_shm->name;
         ShmLocalFree(local_trx_shm);
-        UBRingManager::ReleaseUbrTrxFromMgr(_trx);
+        UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id));
         return rc;
     }
     return UBRING_OK;
@@ -1000,7 +1171,7 @@ RETURN_CODE UBRing::WritevHasEnoughSpace(size_t buf_len)
     return UBRING_OK;
 }
 
-RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time, UbrCloseType close_type)
+RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx)
 {
     if (UNLIKELY(trx == nullptr)) {
         LOG(ERROR) << "Trx close failed, trx is null.";
@@ -1008,16 +1179,17 @@ RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time, UbrC
     }
 
     UbrEventQMsg* local_tx_event_q = (UbrEventQMsg *)trx->ubr_tx.local_tx_event_q.addr;
+    if (UNLIKELY(local_tx_event_q == nullptr)) {
+        LOG(ERROR) << "Trx close failed, local_tx_event_q addr is NULL, trx local name=" << trx->local_shm.name;
+        return UBRING_ERR;
+    }
     if (local_tx_event_q->flag == UBR_STATE_CONNECTED) {
         local_tx_event_q->flag = UBR_STATE_CLOSING;
     }
 
-    if (close_type == UBR_SEND_CLOSE) {
-        DeleteTimerSafe((uint32_t)trx->timer_fd);
-    } else {
-        DeleteTimer((uint32_t)trx->timer_fd);
-    }
-    DeleteTimerSafe((uint32_t)trx->hb_timer_fd);
+    // Non-blocking: may run inside the close callback itself.
+    UbrTimerDel(&trx->close_timer);
+    UbrTimerDel(&trx->hb_timer);
 
     if (local_tx_event_q->flag == UBR_STATE_CLOSING) {
         local_tx_event_q->flag = UBR_STATE_CLOSED;
@@ -1027,9 +1199,9 @@ RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time, UbrC
     return UBRING_OK;
 }
 
-RETURN_CODE UBRing::ClearTrxResource(UbrTrx *trx, uint64_t start_time, UbrCloseType close_type, int op)
+RETURN_CODE UBRing::ClearTrxResource(UbrTrx *trx)
 {
-    RETURN_CODE rc = UbrClearResourceCheck(trx, start_time, close_type);
+    RETURN_CODE rc = UbrClearResourceCheck(trx);
     if (rc != UBRING_OK) {
         return rc;
     }
diff --git a/src/brpc/ubshm/ub_ring.h b/src/brpc/ubshm/ub_ring.h
index 0a02dd2fde..0e47a04d4e 100644
--- a/src/brpc/ubshm/ub_ring.h
+++ b/src/brpc/ubshm/ub_ring.h
@@ -55,12 +55,14 @@ class UBRing : public butil::IReader {
 
     static void *UbrTrxHBCallback(void *args);
 
-    static RETURN_CODE UbrPassiveClearTrx(UbrTrx *trx, int fd, PASSIVE_DISC_TYPE type);
+    static RETURN_CODE UbrPassiveClearTrx(UbrTrx *trx);
 
     static RETURN_CODE UbrAddAsynClearTimer(UbrTrx *trx);
 
     static void *UbrAsynClearCallback(void *args);
 
+    static void *UbrPassiveClearCallback(void *args);
+
     int UbrTrxSend(const void *buf, uint32_t buf_len);
 
     int UbrTrxRecv(void *buf, uint32_t buf_len);
@@ -212,8 +214,8 @@ class UBRing : public butil::IReader {
     void PreWriteAddr(uint8_t *addr, size_t len);
     RETURN_CODE WritevHasEnoughSpace(size_t buf_len);
     RETURN_CODE UbrServerTrxInit(SHM *local_shm, SHM *remote_shm);
-    static RETURN_CODE UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time, UbrCloseType close_type);
-    static RETURN_CODE ClearTrxResource(UbrTrx *trx, uint64_t start_time, UbrCloseType close_type, int op=0);
+    static RETURN_CODE UbrClearResourceCheck(UbrTrx *trx);
+    static RETURN_CODE ClearTrxResource(UbrTrx *trx);
 
     UbrTrx* _trx{nullptr};
 };
diff --git a/src/brpc/ubshm/ub_ring_manager.cpp b/src/brpc/ubshm/ub_ring_manager.cpp
index 64f2434eba..6abb485e8a 100644
--- a/src/brpc/ubshm/ub_ring_manager.cpp
+++ b/src/brpc/ubshm/ub_ring_manager.cpp
@@ -22,6 +22,11 @@
 
 namespace brpc {
 namespace ubring {
+
+// A UbrCleanupCtl reference count of 1 means only the manager anchor is
+// left, i.e. no cleanup callback is in flight for it.
+static constexpr int kAnchoredRefOnly = 1;
+
 DEFINE_int32(ubr_max_managed_num, 1024, "maximum number of managed ubring");
 DEFINE_int32(tail_update_after_read, 8, "Position of the tail update after the read");
 
@@ -54,6 +59,8 @@ RETURN_CODE UBRingManager::UbrMgrDefault()
     g_ubr_mgr.trx_cap = FLAGS_ubr_max_managed_num;
     g_ubr_mgr.trx_mgr_unit_status = nullptr;
     g_ubr_mgr.trx_mgr = nullptr;
+    g_ubr_mgr.trx_mgr_unit_id = nullptr;
+    g_ubr_mgr.trx_mgr_unit_ctl = nullptr;
     return UBRING_OK;
 }
 
@@ -68,8 +75,14 @@ RETURN_CODE UBRingManager::UbrMgrInit() {
     g_ubr_mgr.trx_mgr = (UbrTrx *)malloc(trx_mgr_size);
     size_t trx_mgr_status_size = g_ubr_mgr.trx_cap * sizeof(UbrMgrUnitStatus);
     g_ubr_mgr.trx_mgr_unit_status = (UbrMgrUnitStatus *)malloc(trx_mgr_status_size);
+    size_t trx_mgr_id_size = g_ubr_mgr.trx_cap * sizeof(uint64_t);
+    g_ubr_mgr.trx_mgr_unit_id = (uint64_t *)malloc(trx_mgr_id_size);
+    size_t trx_mgr_ctl_size = g_ubr_mgr.trx_cap * sizeof(UbrCleanupCtl *);
+    g_ubr_mgr.trx_mgr_unit_ctl = (UbrCleanupCtl **)malloc(trx_mgr_ctl_size);
     if (UNLIKELY(g_ubr_mgr.trx_mgr == nullptr ||
-                 g_ubr_mgr.trx_mgr_unit_status == nullptr)) {
+                 g_ubr_mgr.trx_mgr_unit_status == nullptr ||
+                 g_ubr_mgr.trx_mgr_unit_id == nullptr ||
+                 g_ubr_mgr.trx_mgr_unit_ctl == nullptr)) {
         LOG(ERROR) << "Ubr manager memory allocation failed.";
         UbrMgrFini();
         return UBRING_ERR;
@@ -77,15 +90,57 @@ RETURN_CODE UBRingManager::UbrMgrInit() {
 
     memset(g_ubr_mgr.trx_mgr, 0, trx_mgr_size);
     memset(g_ubr_mgr.trx_mgr_unit_status, UBR_MGR_UNIT_FREE, trx_mgr_status_size);
+    memset(g_ubr_mgr.trx_mgr_unit_id, 0, trx_mgr_id_size);
+    memset(g_ubr_mgr.trx_mgr_unit_ctl, 0, trx_mgr_ctl_size);
     LinkInfoInit();
     return UBRING_OK;
 }
 
 void UBRingManager::UbrMgrFini() {
+    // Cancel the pending delayed cleanups and wait for the in-flight ones
+    // (each holds one extra reference) to finish, before the pool memory
+    // they touch is freed. A ctl whose timer is still starting can only be
+    // cancelled in a later round, hence the retry-to-stability loop.
+    bool busy = true;
+    while (busy) {
+        busy = false;
+        {
+            LOCK_GUARD(g_ubr_trx_mgr_mtx);
+            if (g_ubr_mgr.trx_mgr_unit_ctl != nullptr) {
+                for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) {
+                    UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[i];
+                    if (ctl == nullptr) {
+                        continue;
+                    }
+                    if (UbrTimerDel(&ctl->timer) == 0) {
+                        ctl->ReleaseRef();   // timer/callback reference
+                    }
+                    if (ctl->ref.load() > kAnchoredRefOnly) {
+                        busy = true;
+                    }
+                }
+            }
+        }
+        if (busy) {
+            LOG_EVERY_SECOND(INFO) << "UbrMgrFini waits for in-flight cleanups.";
+            usleep(1000);
+        }
+    }
     {
         LOCK_GUARD(g_ubr_trx_mgr_mtx);
+        if (g_ubr_mgr.trx_mgr_unit_ctl != nullptr) {
+            for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) {
+                UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[i];
+                if (ctl != nullptr) {
+                    g_ubr_mgr.trx_mgr_unit_ctl[i] = nullptr;
+                    ctl->ReleaseRef();           // manager reference
+                }
+            }
+        }
         FREE_PTR(g_ubr_mgr.trx_mgr);
         FREE_PTR(g_ubr_mgr.trx_mgr_unit_status);
+        FREE_PTR(g_ubr_mgr.trx_mgr_unit_id);
+        FREE_PTR(g_ubr_mgr.trx_mgr_unit_ctl);
     }
     {
         LOCK_GUARD(g_ubr_listener_mgr_mtx);
@@ -115,10 +170,25 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) {
     for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) {
         if (g_ubr_mgr.trx_mgr_unit_status[i] == UBR_MGR_UNIT_FREE) {
             memset(&g_ubr_mgr.trx_mgr[i], 0, sizeof(UbrTrx));
+            // The explicit re-initialization after memset is deliberate: it
+            // documents the per-acquisition invariants of these fields.
+            g_ubr_mgr.trx_mgr[i].close_timer = nullptr;
+            g_ubr_mgr.trx_mgr[i].hb_timer = nullptr;
+            g_ubr_mgr.trx_mgr[i].cleanup_ctl = nullptr;
+            // Retire the previous acquisition's cleanup control object.
+            UbrCleanupCtl* old_ctl = g_ubr_mgr.trx_mgr_unit_ctl[i];
+            g_ubr_mgr.trx_mgr_unit_ctl[i] = nullptr;
+            if (old_ctl != nullptr) {
+                if (UbrTimerDel(&old_ctl->timer) == 0) {
+                    old_ctl->ReleaseRef();       // timer/callback reference
+                }
+                old_ctl->ReleaseRef();           // manager anchor
+            }
             g_ubr_mgr.trx_mgr_unit_status[i] = UBR_MGR_UNIT_USED;
             *trx = &g_ubr_mgr.trx_mgr[i];
             (*trx)->trx_mgr_index = i;
-            (*trx)->ubr_id = g_ubr_trx_num;
+            ATOMIC_STORE((*trx)->ubr_id, g_ubr_trx_num);
+            g_ubr_mgr.trx_mgr_unit_id[i] = g_ubr_trx_num;
             (*trx)->close_state = UBR_CLOSE_FIRST;
             (*trx)->close_cnt = MAX_CLOSE_COUNT;
             ++g_ubr_mgr.trx_num;
@@ -130,17 +200,12 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) {
     return UBRING_ERR;
 }
 
-RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx) {
+RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx,
+                                                uint64_t expect_ubr_id) {
     if (UNLIKELY(trx == nullptr)) {
         LOG(ERROR) << "Release trx failed, trx is null.";
         return UBRING_ERR;
     }
-
-    trx->local_shm.addr = nullptr;
-    trx->ubr_tx.local_tx_event_q.addr = nullptr;
-    trx->ubr_tx.local_data_status_q.addr = nullptr;
-    trx->ubr_rx.local_rx_event_q.addr = nullptr;
-    trx->ubr_rx.remote_data_status_q.addr = nullptr;
     if (UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) {
         LOG(ERROR) << "Release trx failed, trx_mgr is null.";
         return UBRING_ERR;
@@ -153,16 +218,82 @@ RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx) {
         return UBRING_OK;
     }
 
+    if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id)) {
+        // The slot was released and acquired again meanwhile; the stale
+        // caller must not touch the new occupant.
+        LOG(WARNING) << "Release stale trx refused, name=" << trx->local_shm.name;
+        return UBRING_OK;
+    }
+
     if (g_ubr_mgr.trx_num == 0) {
         LOG(ERROR) << "Release trx failed, trx number is 0.";
         return UBRING_ERR;
     }
 
+    // Mutate the trx only after the generation check passed.
+    trx->local_shm.addr = nullptr;
+    trx->ubr_tx.local_tx_event_q.addr = nullptr;
+    trx->ubr_tx.local_data_status_q.addr = nullptr;
+    trx->ubr_rx.local_rx_event_q.addr = nullptr;
+    trx->ubr_rx.remote_data_status_q.addr = nullptr;
     g_ubr_mgr.trx_mgr_unit_status[idx] = UBR_MGR_UNIT_FREE;
     --g_ubr_mgr.trx_num;
     return UBRING_OK;
 }
 
+UbrCleanupCtl* UBRingManager::SnapshotUnitCleanupCtl(uint32_t idx) {
+    LOCK_GUARD(g_ubr_trx_mgr_mtx);
+    if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || idx >= g_ubr_mgr.trx_cap)) {
+        return nullptr;
+    }
+    UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[idx];
+    if (ctl != nullptr) {
+        ctl->ref.fetch_add(1);               // snapshot reference
+    }
+    return ctl;
+}
+
+bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) {
+    LOCK_GUARD(g_ubr_trx_mgr_mtx);
+    if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_id == nullptr ||
+                 g_ubr_mgr.trx_mgr_unit_status == nullptr ||
+                 idx >= g_ubr_mgr.trx_cap)) {
+        return false;
+    }
+    return g_ubr_mgr.trx_mgr_unit_status[idx] == UBR_MGR_UNIT_USED &&
+           g_ubr_mgr.trx_mgr_unit_id[idx] == expect_ubr_id;
+}
+
+bool UBRingManager::TryPublishUnitCleanupCtl(uint32_t idx,
+                                             uint64_t expect_ubr_id,
+                                             UbrCleanupCtl *ctl) {
+    LOCK_GUARD(g_ubr_trx_mgr_mtx);
+    if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr ||
+                 g_ubr_mgr.trx_mgr_unit_status == nullptr ||
+                 g_ubr_mgr.trx_mgr_unit_id == nullptr ||
+                 idx >= g_ubr_mgr.trx_cap ||
+                 g_ubr_mgr.trx_mgr_unit_status[idx] != UBR_MGR_UNIT_USED ||
+                 g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id ||
+                 g_ubr_mgr.trx_mgr_unit_ctl[idx] != nullptr)) {
+        return false;                        // released / reused / already anchored
+    }
+    ctl->ref.fetch_add(1);                   // manager anchor reference
+    g_ubr_mgr.trx_mgr_unit_ctl[idx] = ctl;
+    return true;
+}
+
+bool UBRingManager::DetachUnitCleanupCtl(uint32_t idx, UbrCleanupCtl *ctl) {
+    LOCK_GUARD(g_ubr_trx_mgr_mtx);
+    if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr ||
+                 idx >= g_ubr_mgr.trx_cap ||
+                 g_ubr_mgr.trx_mgr_unit_ctl[idx] != ctl)) {
+        return false;
+    }
+    g_ubr_mgr.trx_mgr_unit_ctl[idx] = nullptr;
+    ctl->ReleaseRef();                           // manager reference
+    return true;
+}
+
 void UBRingManager::LinkInfoInit(void) {
 
     size_t link_info_mgr_size = FLAGS_ubr_max_managed_num * sizeof(UbrLinkInfo);
@@ -255,7 +386,7 @@ int32_t UBRingManager::UbEventCallback(const char *shm_name)
             ++g_ub_event_cnt;
             int fd = (int)g_ubr_mgr.trx_mgr[i].local_shm.fd;
             LOG(WARNING) << "Ub event callback, the fd of the faulty link is " << fd;
-            return UBRing::UbrPassiveClearTrx(&g_ubr_mgr.trx_mgr[i], fd, UBR_UB_EVENT);
+            return UBRing::UbrPassiveClearTrx(&g_ubr_mgr.trx_mgr[i]);
         }
     }
     return UBRING_ERR;
diff --git a/src/brpc/ubshm/ub_ring_manager.h b/src/brpc/ubshm/ub_ring_manager.h
index e2afa324c7..131a749ef7 100644
--- a/src/brpc/ubshm/ub_ring_manager.h
+++ b/src/brpc/ubshm/ub_ring_manager.h
@@ -34,6 +34,8 @@ typedef struct TagUbrMgr {
     uint32_t trx_cap;
     UbrTrx *trx_mgr;
     UbrMgrUnitStatus *trx_mgr_unit_status;
+    uint64_t *trx_mgr_unit_id;
+    UbrCleanupCtl **trx_mgr_unit_ctl;
 } UbrMgr;
 
 typedef struct TagUbrLinkInfo {
@@ -63,7 +65,34 @@ class UBRingManager {
 
     static RETURN_CODE AcquireUbrTrxFromMgr(UbrTrx **trx);
 
-    static RETURN_CODE ReleaseUbrTrxFromMgr(UbrTrx *trx);
+    // Release the pool slot of `trx'. `expect_ubr_id' must be snapshotted
+    // from trx->ubr_id before the caller started releasing the trx: a
+    // release racing a reuse of the same slot is refused.
+    static RETURN_CODE ReleaseUbrTrxFromMgr(UbrTrx *trx,
+                                            uint64_t expect_ubr_id);
+
+    // Snapshot the cleanup control object of a pool slot. The caller
+    // receives a reference and must ReleaseRef it on every exit path.
+    static UbrCleanupCtl* SnapshotUnitCleanupCtl(uint32_t idx);
+
+    // Under the manager lock, confirm the pool slot is still used by the
+    // given generation (not released or reused meanwhile).
+    static bool IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id);
+
+    // Atomically (under the manager lock) anchor `ctl' to the pool slot,
+    // but only while the slot is still used by the given generation and
+    // carries no other cleanup control object. On success the ctl gains
+    // the manager anchor reference (released when the anchor is detached
+    // or the slot is retired); returns false -- leaving the slot and the
+    // reference counts untouched -- when the trx was released, reused, or
+    // a cleanup is already anchored.
+    static bool TryPublishUnitCleanupCtl(uint32_t idx, uint64_t expect_ubr_id,
+                                         UbrCleanupCtl *ctl);
+
+    // Detach `ctl' from the pool slot if it is still anchored there.
+    // Returns true when the detach happened (the manager reference was
+    // released).
+    static bool DetachUnitCleanupCtl(uint32_t idx, UbrCleanupCtl *ctl);
 
     static void LinkInfoInit(void);
     static void LinkInfoFini(void);
diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h
index b4e89645ff..6a947d2d78 100644
--- a/src/brpc/ubshm/ubr_trx.h
+++ b/src/brpc/ubshm/ubr_trx.h
@@ -24,6 +24,7 @@
 #include "brpc/ubshm/common/common.h"
 #include "brpc/ubshm/common/thread_lock.h"
 #include "brpc/ubshm/ubr_msg.h"
+#include "brpc/ubshm/timer/timer_mgr.h"
 
 /* +----------------------------------------------------------------------------+
    │                                 UbrTrx shm                                 │
@@ -57,17 +58,22 @@ typedef enum {
     UBR_STATE_CLOSED
 } EventQState;
 
-typedef enum {
-    UBR_SEND_CLOSE,
-    UBR_CALL_BACK_CLOSE
-} UbrCloseType;
-
 typedef enum {
     UBR_CLOSE_FIRST,
     UBR_CLOSE_SECOND,
     UBR_CLOSE_END
 } UbrCloseCount;
 
+// Ownership of the final delayed cleanup of a trx. Exactly one side (the
+// delayed-clear callback or a force close) may run it. The state lives in
+// a per-acquisition heap control object (UbrCleanupCtl) anchored by the
+// trx manager, so it survives the pool slot reuse that wipes UbrTrx.
+typedef enum {
+    UBR_CLEANUP_PENDING = 1,                     // cleanup timer scheduled
+    UBR_CLEANUP_RUNNING = 2,                     // cleanup owned, in progress
+    UBR_CLEANUP_DONE = 3                         // cleanup finished
+} UbrCleanupState;
+
 typedef enum {
     UDP_TRX,
     TCP_TRX,
@@ -104,11 +110,11 @@ typedef struct TagUbrTx {
     UbrAddrInfo remote_rx_event_q;
     UbrAddrInfo local_data_status_q;
     UbrAddrInfo local_tx_event_q;
-    uint64_t out_io_id;
+    AtomicUintFast64 out_io_id;
     uint32_t write_pos;
     uint32_t capacity;
     UbrMsgFormat local_msg_space;
-    uint32_t hb_retry_cnt;
+    int32_t hb_retry_cnt;
     uint32_t ep_last_cap;
     volatile EventQState trx_state;
 } UbrTx;
@@ -118,7 +124,7 @@ typedef struct TagUbrRx {
     UbrAddrInfo local_rx_event_q;
     UbrAddrInfo remote_data_status_q;
     UbrAddrInfo remote_tx_event_q;
-    uint64_t in_io_id;
+    AtomicUintFast64 in_io_id;
     uint32_t read_pos;
     uint32_t capacity;
     uint32_t deal_msg_num;
@@ -127,17 +133,47 @@ typedef struct TagUbrRx {
     volatile EventQState trx_state;
 } UbrRx;
 
+struct TagUbrTrx;
+
+// Control object of one delayed cleanup, created when the cleanup is
+// scheduled and destroyed when the owning pool slot is acquired again.
+// It is anchored by the trx manager (not by the reusable UbrTrx memory),
+// so the ownership claim and the completion signal stay valid across the
+// release/reuse of the trx that owns the cleanup.
+struct UbrCleanupCtl {
+    TagUbrTrx* trx;                              // immutable
+    uint64_t ubr_id;                             // immutable generation
+    AtomicInt state;                             // UbrCleanupState
+    UbrTimerId timer;                            // delayed clear timer
+    // References: timer/callback (held from a successful schedule until
+    // the callback fully returned, or released by whoever cancels the
+    // timer before it fires) + starter (until the schedule call
+    // completes) + manager anchor (taken when the ctl is anchored to the
+    // pool slot, released when the anchor is detached or retired).
+    AtomicInt ref;
+
+    void ReleaseRef() {
+        if (ref.fetch_sub(1) == 1) {
+            delete this;
+        }
+    }
+};
+
 typedef struct TagUbrTrx {
     UbrTx ubr_tx;
     UbrRx ubr_rx;
-    uint64_t ubr_id;
+    AtomicUintFast64 ubr_id;
     uint32_t trx_mgr_index;
     UbrTrxType type;
     SHM local_shm;
     SHM remote_shm;
-    int timer_fd;
-    int hb_timer_fd;
-    int clear_timer_fd;
+    UbrTimerId close_timer;
+    UbrTimerId hb_timer;
+    UbrCleanupCtl* cleanup_ctl;
+    // Last io ids seen by the close-check timer, used to reset its
+    // back-off polling interval when the link has traffic.
+    uint64_t close_chk_in_io_id;
+    uint64_t close_chk_out_io_id;
     AtomicInt close_cnt;
     AtomicInt close_state;
 } UbrTrx;
@@ -152,11 +188,7 @@ typedef struct TagUbrLinkLock {
     FileLock* file_lock;
 } UbrLinkLock;
 
-typedef enum {
-    UBR_UB_EVENT,
-    UBR_HEARTBEAT,
-}PASSIVE_DISC_TYPE;
-
 }
 }
-#endif //BRPC_UBR_TRX_H
\ No newline at end of file
+#endif //BRPC_UBR_TRX_H
+
diff --git a/test/brpc_ubring_unittest.cpp b/test/brpc_ubring_unittest.cpp
index d3d0a7ce98..ce9bbb062c 100644
--- a/test/brpc_ubring_unittest.cpp
+++ b/test/brpc_ubring_unittest.cpp
@@ -37,6 +37,7 @@ DECLARE_int32(ub_disconnect_timeout_s);
 DECLARE_int32(ub_connect_timeout_s);
 DECLARE_int32(ub_hb_timer_interval_s);
 DECLARE_int32(ub_event_queue_timer_interval_us);
+DECLARE_int32(ub_event_queue_timer_interval_max_us);
 DECLARE_int32(ub_flying_io_timeout_s);
 
 extern bool g_skip_ub_init;
@@ -157,6 +158,7 @@ TEST(UBRingConfigurationTest, time_flags_include_units_and_expected_defaults) {
         {"ub_connect_timeout_s", "_s", "seconds", "1"},
         {"ub_hb_timer_interval_s", "_s", "seconds", "5"},
         {"ub_event_queue_timer_interval_us", "_us", "microseconds", "100"},
+        {"ub_event_queue_timer_interval_max_us", "_us", "microseconds", "10000"},
         {"ub_flying_io_timeout_s", "_s", "seconds", "5"},
     };
 
@@ -176,6 +178,7 @@ TEST(UBRingConfigurationTest, time_flags_include_units_and_expected_defaults) {
     EXPECT_EQ(1, brpc::ubring::FLAGS_ub_connect_timeout_s);
     EXPECT_EQ(5, brpc::ubring::FLAGS_ub_hb_timer_interval_s);
     EXPECT_EQ(100, brpc::ubring::FLAGS_ub_event_queue_timer_interval_us);
+    EXPECT_EQ(10000, brpc::ubring::FLAGS_ub_event_queue_timer_interval_max_us);
     EXPECT_EQ(5, brpc::ubring::FLAGS_ub_flying_io_timeout_s);
     EXPECT_EQ(100U * USEC_TO_NSEC,
               static_cast<uint32_t>(