diff --git a/docs/cn/ubring.md b/docs/cn/ubring.md
index f4c7c0b841..182de6cba6 100644
--- a/docs/cn/ubring.md
+++ b/docs/cn/ubring.md
@@ -187,7 +187,7 @@ UBRing 架构包含以下组件：
 
 ### 定时器管理
 
-UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删，此时句柄会保留到该回调返回，因此并发的外部 `UbrTimerDelAndWait` 一定能等到它；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。进入关闭处理时会带上调用方看到的代际，并在 manager 锁内原子认领槽位，因此按旧快照发起的关闭（例如共享内存故障事件回调）不会误伤释放后被复用的新连接。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
+UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删，此时句柄会保留到该回调返回，因此并发的外部 `UbrTimerDelAndWait` 一定能等到它；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。进入关闭处理时会带上调用方看到的代际，并在 manager 锁内原子认领槽位，因此按旧快照发起的关闭（例如共享内存故障事件回调）不会误伤释放后被复用的新连接。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。阻塞的共享内存清理不在这里进行：延迟清理与待回收映射的回调只仲裁清理所有权，并把任务投递给专用的清理 worker bthread（`ub_cleanup_worker.cpp`），由它把 UBS SDK 的 `unmap`/`deallocate` 调用移出定时线程。对一次性定时器，`UbrTimerDel` 返回 0 还额外保证 facade 已不再访问句柄槽位所在的存储（它会等待被定时线程派发但尚未竞争槽位的 wrapper 退出竞争），因此调用方可以释放承载该槽位的对象；延迟清理路径正是据此释放 `UbrCleanupCtl`。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
 
 ## 参考资料
 
diff --git a/docs/en/ubring.md b/docs/en/ubring.md
index 9247360e04..8cb64df6dc 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 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; a periodic callback that deletes its own timer keeps its handle anchored until the callback returns, so a concurrent `UbrTimerDelAndWait` from a teardown path always waits for it. 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. Entering a close carries the generation the caller saw and claims the pool slot atomically under the manager lock, so a close started from a stale snapshot (the shared-memory fault event, for instance) cannot hit a connection that reused the slot in the meantime. 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.
+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; a periodic callback that deletes its own timer keeps its handle anchored until the callback returns, so a concurrent `UbrTimerDelAndWait` from a teardown path always waits for it. 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. Entering a close carries the generation the caller saw and claims the pool slot atomically under the manager lock, so a close started from a stale snapshot (the shared-memory fault event, for instance) cannot hit a connection that reused the slot in the meantime. Timer callbacks run on the process-wide bthread timer thread and must return quickly. Blocking shared-memory cleanup is therefore not done there: the delayed-clear and pending-unmap callbacks only settle the cleanup ownership and post a job to a dedicated cleanup worker bthread (`ub_cleanup_worker.cpp`), which runs the UBS SDK `unmap`/`deallocate` calls off the timer thread. For a one-shot timer, a `UbrTimerDel` that returns 0 additionally guarantees that the facade is done with the storage the handle slot lives in -- it waits out a wrapper the timer thread already dispatched -- so the caller may free the object that holds the slot; the delayed-clear path frees `UbrCleanupCtl` that way. 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/shm/shm_ubs.cpp b/src/brpc/ubshm/shm/shm_ubs.cpp
index aa4086619d..699c26dd44 100644
--- a/src/brpc/ubshm/shm/shm_ubs.cpp
+++ b/src/brpc/ubshm/shm/shm_ubs.cpp
@@ -24,6 +24,7 @@
 #include <gflags/gflags.h>
 #include "butil/scoped_lock.h"
 #include "brpc/ubshm/timer/timer_mgr.h"
+#include "brpc/ubshm/ub_cleanup_worker.h"
 #include "brpc/ubshm/common/common.h"
 #include "brpc/ubshm/shm/shm_def.h"
 #include "brpc/ubshm/ub_ring_manager.h"
@@ -49,6 +50,11 @@ DEFINE_int32(ub_flying_io_timeout_s, 5,
 char g_region_name[MAX_REGION_NAME_DESC_LENGTH] = {0};
 butil::atomic<UbrTimerId> g_shm_timer_id(nullptr);
 ShmList *g_shm_list = nullptr;
+// True while a drain step of g_shm_list is queued on (or running in) the
+// cleanup worker. It bounds the job backlog: the periodic timer only posts a
+// new step when the previous one finished, and the worker clears the flag at
+// the end of every step, successful or not.
+static butil::atomic<bool> g_shm_drain_pending(false);
 // Set by UbsShmFini, cleared by UbsShmInit: ShmMgrFini may run more than once,
 // and the SDK finalize plus the shm list teardown must happen exactly once per
 // init.
@@ -434,13 +440,38 @@ void *UbsShmCallback(void* args, uint64_t)
         return nullptr;
     }
 
-    // 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.
+    // The timer thread runs every timer in the process, so it must not call
+    // into the SDK: post the drain to the cleanup worker. At most one step may
+    // be in flight, otherwise a slow SDK call would pile up jobs for the same
+    // list head.
+    if (g_shm_drain_pending.exchange(true)) {
+        return nullptr;
+    }
+    if (BAIDU_UNLIKELY(!UbrCleanupWorker::PostShmDrain(shm_list, UbsShmDrainNode))) {
+        // The worker could not be started: keep the pre-worker behaviour and
+        // drain here rather than leaving the nodes in the list forever.
+        LOG(ERROR) << "Cleanup worker unavailable, draining shm on the timer thread.";
+        UbsShmDrainNode(shm_list);
+    }
+    return nullptr;
+}
+
+void UbsShmDrainNode(ShmList *shm_list)
+{
+    if (BAIDU_UNLIKELY(shm_list == nullptr)) {
+        LOG(ERROR) << "Shm list is null.";
+        g_shm_drain_pending.store(false);
+        return;
+    }
+
+    // Drain one node per step and keep the SDK calls outside the lock, so a
+    // slow daemon cannot stall other users of the list.
     SHM shm;
     {
         BAIDU_SCOPED_LOCK(shm_list->shm_lock);
         if (shm_list->head == nullptr) {
-            return nullptr;
+            g_shm_drain_pending.store(false);
+            return;
         }
         shm = shm_list->head->shm;
     }
@@ -448,16 +479,18 @@ void *UbsShmCallback(void* args, uint64_t)
         LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
         BAIDU_SCOPED_LOCK(shm_list->shm_lock);
         DeleteShmToList(shm_list);
-        return nullptr;
+        g_shm_drain_pending.store(false);
+        return;
     }
 
     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
+        if (ret != UBSM_ERR_NET) {
+            LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
         }
-        LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
-        return nullptr;                  // node stays at head, retried
+        // The node stays at the head: the next timer fire retries it.
+        g_shm_drain_pending.store(false);
+        return;
     }
 
     {
@@ -469,7 +502,7 @@ void *UbsShmCallback(void* args, uint64_t)
     if (ret != UBSM_OK) {
         LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
     }
-    return nullptr;
+    g_shm_drain_pending.store(false);
 }
 
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
@@ -529,8 +562,11 @@ RETURN_CODE InitShmTimer(ShmList **shm_list)
 
 RETURN_CODE DestroyShmTimer(ShmList **shm_list)
 {
-    // Wait out a possibly running UbsShmCallback before tearing shm_list down.
+    // Wait out a possibly running UbsShmCallback, then the cleanup jobs it
+    // posted: the worker may be inside the SDK calls of a node that is freed
+    // right below, and must not race ubsmem_finalize either.
     UbrTimerDelAndWait(&g_shm_timer_id);
+    UbrCleanupWorker::DrainAndWait();
     if (BAIDU_UNLIKELY(shm_list == nullptr)) {
         LOG(ERROR) << "Shm list handle is null.";
         return UBRING_ERR;
diff --git a/src/brpc/ubshm/shm/shm_ubs.h b/src/brpc/ubshm/shm/shm_ubs.h
index 02ed51cd46..562e1f52c2 100644
--- a/src/brpc/ubshm/shm/shm_ubs.h
+++ b/src/brpc/ubshm/shm/shm_ubs.h
@@ -42,6 +42,11 @@ RETURN_CODE UbsShmLocalMmap(SHM *shm, int prot);
 void UbsMemLoggerPrint(int level, const char *msg);
 
 void *UbsShmCallback(void* args, uint64_t gen);
+// One drain step of the pending-unmap list: unmap the head node and deallocate
+// it. Called by the cleanup worker (and inline by UbsShmCallback when that
+// worker cannot be started), never by the timer thread directly. The head node
+// stays in the list when the unmap fails, so a later timer fire retries it.
+void UbsShmDrainNode(ShmList *shm_list);
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list);
 RETURN_CODE InitShmTimer(ShmList **shm_list);
 // Takes the handle so it can be cleared after the list is freed: a caller that
diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp
index 344c365a9f..8eb864dd00 100644
--- a/src/brpc/ubshm/timer/timer_mgr.cpp
+++ b/src/brpc/ubshm/timer/timer_mgr.cpp
@@ -71,6 +71,14 @@ struct UbrTimerTask {
     butil::atomic<int> ref;
     butil::atomic<bool> join_pending;            // a DelAndWait is waiting
     butil::atomic<bool> done;                    // refs hit zero, joiner frees
+    // Set by whoever settles the handle slot: the one-shot wrapper (right
+    // after its slot CAS, before it runs the user callback), a deleter whose
+    // bthread_timer_del cancelled the task before dispatch, or the kDead
+    // path. Once true, no facade code -- including the dispatched wrapper --
+    // will ever touch `*slot` again, so the object storing the slot may be
+    // freed. The scheduled wrapper is kept alive by its own schedule
+    // reference while a deleter waits on this flag.
+    butil::atomic<bool> slot_retired;
 };
 
 namespace {
@@ -160,6 +168,13 @@ void UbrTimerOnFire(void* p) {
     // the slot is the single arbiter.
     UbrTimerId expected = task;
     const bool owned = task->slot->compare_exchange_strong(expected, nullptr);
+    // The slot storage is now retired: whatever the competition above
+    // decided, this wrapper will not dereference `slot' again. A one-shot
+    // UbrTimerDel that won the exchange (or that cancelled a task which had
+    // not been dispatched at all) is waiting for this store before it lets
+    // the caller free the storage, so it must happen before the potentially
+    // long user callback.
+    task->slot_retired.store(true);
     if (owned) {
         task->cb(task->arg, task->gen);
     }
@@ -208,6 +223,7 @@ RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t delay_u
     task->ref.store(3);                          // owner + schedule + starter
     task->join_pending.store(false);
     task->done.store(false);
+    task->slot_retired.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.
@@ -293,19 +309,41 @@ int UbrTimerDel(butil::atomic<UbrTimerId>* slot) {
         bthread_usleep(1000);
     }
     if (task->state.load() == kDead) {
+        // The wrapper never ran and never will: this call is the one that
+        // settles the slot storage.
+        task->slot_retired.store(true);
         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) {
+        // Cancelled before dispatch: the timer thread will never call the
+        // wrapper, so nothing else can retire the slot storage.
+        task->slot_retired.store(true);
         ReleaseRef(task);                        // schedule: cancelled before dispatch
-    }                                            // ==1: dispatched, OnFire (owned==false)
-                                                 // releases it
+    } else if (!task->periodic) {
+        // One-shot already dispatched: the wrapper may be between its dispatch
+        // (which made bthread_timer_del return 1) and its slot CAS, so it
+        // still has to touch the storage before it returns. Wait it out here,
+        // while this caller still owns the task through the owner reference,
+        // so that returning 0 really means "the storage is not used anymore".
+        // The wrapper retires the slot before calling any user code and takes
+        // no lock doing so, hence this wait cannot deadlock, and it is bounded
+        // by the OS scheduling of the timer thread. A periodic task needs no
+        // such wait: its slot storage is the pooled UbrTrx, which the callers
+        // of the non-blocking delete never free (only UbrMgrFini does, after
+        // UbrTimerDelAndWait), and its retire-slot load happens after the user
+        // callback -- which may take the very lock this caller holds.
+        while (!task->slot_retired.load()) {
+            bthread_usleep(kTimerPollIntervalUs);
+        }
+    }                                            // ==1 periodic: dispatched/running;
+                                                 // OnFire releases the schedule ref
     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.
+                    // the callback will not run and the slot storage is unused.
+                    // For a periodic timer, future rearming is stopped, but an
+                    // already dispatched or running callback may still complete.
 }
 
 void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {
diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h
index b8d3676ab6..1c35a22499 100644
--- a/src/brpc/ubshm/timer/timer_mgr.h
+++ b/src/brpc/ubshm/timer/timer_mgr.h
@@ -64,16 +64,27 @@ RETURN_CODE UbrTimerStartPeriodic(butil::atomic<UbrTimerId>* slot,
                                   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.
+// This function does not wait for an already running user 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 one-shot timer, the callback will not run. This call additionally
+//   waits out a wrapper that the timer thread already dispatched: such a
+//   wrapper may have been preempted between its dispatch and the moment it
+//   competes for the slot, and it still has to touch the slot once (to lose
+//   that competition) before it can return. When this call returns 0, the
+//   facade is therefore completely done with the storage `slot` points to,
+//   and the caller may free the object that stores the slot -- which is what
+//   the delayed-clear path does with UbrCleanupCtl. The wait only spins on a
+//   task-internal flag: the wrapper takes no lock and calls no user code
+//   while retiring the slot, so a caller holding any ubring lock can still
+//   wait, and no timer callback can be joined from its own dispatch.
 // - 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.
-//   A return of 0 therefore never means "the handle is free again".
+//   dispatched or running callback may still execute once more, and the task
+//   stays anchored in the slot until that callback returns. Callers must not
+//   reclaim resources reachable from `arg` based on this return alone; use
+//   UbrTimerDelAndWait when teardown needs to wait for callbacks. A return of
+//   0 therefore never means "the handle is free again".
 // A periodic callback that deletes its own timer falls into the periodic case
 // as well, but only marks it stopped: the handle slot stays anchored until the
 // callback returns, so a concurrent UbrTimerDelAndWait on the same slot really
diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp
new file mode 100644
index 0000000000..151862aa5c
--- /dev/null
+++ b/src/brpc/ubshm/ub_cleanup_worker.cpp
@@ -0,0 +1,153 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <pthread.h>
+#include <stdint.h>
+#include "butil/atomicops.h"
+#include "butil/containers/mpsc_queue.h"
+#include "butil/logging.h"
+#include "butil/scoped_lock.h"
+#include "bthread/bthread.h"
+#include "brpc/ubshm/ub_cleanup_worker.h"
+#include "brpc/ubshm/ub_ring_manager.h"
+
+namespace brpc {
+namespace ubring {
+
+namespace {
+
+// Cleanup latency is tolerated in seconds (it only delays releasing memory of
+// an already dead link), so the idle worker polls instead of spinning and the
+// module needs no extra wakeup primitive.
+constexpr uint64_t kCleanupIdlePollIntervalUs = 100 * 1000;
+
+butil::atomic<bool> g_worker_started(false);
+// Set when the worker bthread could not be created. Terminal: posting then
+// fails forever, and every caller falls back to running its work inline -- the
+// behaviour from before the worker existed. Never cleared, so no job can ever
+// be queued behind a worker that will not run.
+butil::atomic<bool> g_worker_unavailable(false);
+// Posted-but-not-finished jobs. A poster increments it before enqueueing and
+// the worker decrements it after the job returned, so DrainAndWait can wait
+// for quiescence without observing the queue itself.
+butil::atomic<int> g_cleanup_inflight(0);
+butil::MPSCQueue<UbrCleanupJob> g_cleanup_queue;
+// Serializes the posters (never taken by the worker) around the lazy start and
+// the enqueue; see Post().
+pthread_mutex_t g_worker_mtx = PTHREAD_MUTEX_INITIALIZER;
+
+void RunTrxCleanup(const UbrCleanupJob& job) {
+    UbrCleanupCtl* ctl = job.ctl;
+    // The slot may have been released and reused while the job was queued; the
+    // cleanup body must then be skipped, but the control object still has to
+    // be settled (exactly as the timer callback did before the offload).
+    if (UBRingManager::IsUbrTrxSlotUsed(job.trx->trx_mgr_index, job.ubr_id)) {
+        job.work(job.trx, job.ubr_id);
+    }
+    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+    ctl->ReleaseRef();                       // timer/callback reference
+}
+
+void* UbrCleanupWorkerMain(void*) {
+    UbrCleanupJob job;
+    for (;;) {
+        if (!g_cleanup_queue.Dequeue(job)) {
+            bthread_usleep(kCleanupIdlePollIntervalUs);
+            continue;
+        }
+        if (job.type == UbrCleanupJob::SHM_DRAIN) {
+            job.shm_step(job.shm_list);
+        } else {
+            RunTrxCleanup(job);
+        }
+        g_cleanup_inflight.fetch_sub(1);
+    }
+    return nullptr;
+}
+
+bool Post(const UbrCleanupJob& job) {
+    // Serialize the posters so that "the worker is started" and "a job was
+    // enqueued" cannot be decided by different threads at the same time.
+    // Without this, a poster could observe g_worker_started == true (set by
+    // the thread that is starting the bthread) and enqueue a job while that
+    // start is about to fail: g_worker_unavailable is terminal, so the job
+    // would never be dequeued -- its UbrCleanupCtl reference would leak and
+    // g_cleanup_inflight would never return to zero, hanging every later
+    // DrainAndWait on a teardown path. Posting only happens on cleanup paths,
+    // so the mutex is free in practice. The worker itself never takes it.
+    BAIDU_SCOPED_LOCK(g_worker_mtx);
+    if (g_worker_unavailable.load()) {
+        return false;
+    }
+    if (!g_worker_started.load()) {
+        bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
+        bthread_attr_set_name(&attr, "UBCleanup");
+        bthread_t tid;
+        if (BAIDU_UNLIKELY(bthread_start_background(
+                &tid, &attr, UbrCleanupWorkerMain, nullptr) != 0)) {
+            LOG(ERROR) << "Fail to start the ubring cleanup worker bthread.";
+            // Terminal: never queue a job behind a worker that cannot run, so
+            // UbrCleanupWorker::DrainAndWait can never observe an orphaned job.
+            g_worker_unavailable.store(true);
+            return false;
+        }
+        g_worker_started.store(true);
+    }
+    // Increment before enqueueing so a concurrent DrainAndWait can never
+    // observe quiescence while this job is still on its way to the queue.
+    g_cleanup_inflight.fetch_add(1);
+    g_cleanup_queue.Enqueue(job);
+    return true;
+}
+
+}  // namespace
+
+bool UbrCleanupWorker::PostShmDrain(ShmList* shm_list,
+                                    void (*shm_step)(ShmList* shm_list)) {
+    if (BAIDU_UNLIKELY(shm_list == nullptr || shm_step == nullptr)) {
+        return false;
+    }
+    UbrCleanupJob job;
+    job.type = UbrCleanupJob::SHM_DRAIN;
+    job.shm_list = shm_list;
+    job.shm_step = shm_step;
+    return Post(job);
+}
+
+bool UbrCleanupWorker::PostTrxCleanup(UbrTrx* trx, uint64_t ubr_id,
+                                      void (*work)(UbrTrx*, uint64_t),
+                                      UbrCleanupCtl* ctl) {
+    if (BAIDU_UNLIKELY(trx == nullptr || work == nullptr || ctl == nullptr)) {
+        return false;
+    }
+    UbrCleanupJob job;
+    job.type = UbrCleanupJob::TRX_CLEANUP;
+    job.trx = trx;
+    job.ubr_id = ubr_id;
+    job.work = work;
+    job.ctl = ctl;
+    return Post(job);
+}
+
+void UbrCleanupWorker::DrainAndWait() {
+    while (g_cleanup_inflight.load() > 0) {
+        bthread_usleep(1000);
+    }
+}
+
+}  // namespace ubring
+}  // namespace brpc
diff --git a/src/brpc/ubshm/ub_cleanup_worker.h b/src/brpc/ubshm/ub_cleanup_worker.h
new file mode 100644
index 0000000000..005d973fc0
--- /dev/null
+++ b/src/brpc/ubshm/ub_cleanup_worker.h
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_UB_CLEANUP_WORKER_H
+#define BRPC_UB_CLEANUP_WORKER_H
+
+#include <stdint.h>
+#include "brpc/ubshm/common/common.h"
+#include "brpc/ubshm/shm/shm_def.h"
+
+namespace brpc {
+namespace ubring {
+
+struct TagUbrTrx;
+typedef struct TagUbrTrx UbrTrx;
+struct UbrCleanupCtl;
+
+// One unit of ubring cleanup work.
+//
+// The process-wide bthread timer thread runs every timer in the process (RPC
+// timeouts, backup requests, bthread_usleep wakeups, ubring heartbeats and
+// close checks), so a callback that calls into the UBS SDK -- which may wait
+// on a network round trip to the SDK daemon -- delays all of them. The timer
+// callbacks therefore only arbitrate ownership and post a job here; the
+// blocking part runs on the dedicated cleanup worker.
+struct UbrCleanupJob {
+    enum Type {
+        SHM_DRAIN = 0,      // release one pending shared-memory unmap
+        TRX_CLEANUP = 1     // run the delayed cleanup of one trx generation
+    };
+
+    Type type = SHM_DRAIN;
+
+    // SHM_DRAIN: `shm_step' performs the blocking SDK calls and settles the
+    // producer's retry bookkeeping. It is a function pointer so this worker
+    // stays independent of the shm backend, and so its scheduling can be
+    // unit-tested without the SDK.
+    void (*shm_step)(ShmList* shm_list) = nullptr;
+    ShmList* shm_list = nullptr;
+
+    // TRX_CLEANUP: the job takes over the timer/callback reference the
+    // scheduling callback held on `ctl' and releases it when done. `work' is
+    // the cleanup body (used by UbrDoAsynClearWork / UbrDoPassiveClearWork)
+    // and only runs if the slot still belongs to `ubr_id'.
+    UbrTrx* trx = nullptr;
+    uint64_t ubr_id = 0;
+    void (*work)(UbrTrx* trx, uint64_t ubr_id) = nullptr;
+    UbrCleanupCtl* ctl = nullptr;
+};
+
+// A lazily started, process-wide bthread that drains UbrCleanupJob's. It is
+// never stopped during a normal run: after a ShmMgrFini cycle the worker only
+// needs to be quiescent (DrainAndWait) before the shared memory it may touch
+// is finalized, and an idle worker performs no SDK call at all. Posting is
+// serialized, so a job can never be queued behind a worker start that failed.
+class UbrCleanupWorker {
+public:
+    // Post a shared-memory drain step. Returns false when the worker cannot be
+    // started, in which case the caller must run `shm_step' inline.
+    static bool PostShmDrain(ShmList* shm_list, void (*shm_step)(ShmList* shm_list));
+
+    // Post a delayed trx cleanup, transferring the timer/callback reference of
+    // `ctl'. Returns false when the worker cannot be started, in which case
+    // the caller keeps the reference and must run the cleanup inline.
+    static bool PostTrxCleanup(UbrTrx* trx, uint64_t ubr_id,
+                               void (*work)(UbrTrx* trx, uint64_t ubr_id),
+                               UbrCleanupCtl* ctl);
+
+    // Wait until every posted job finished. Bounded by one in-flight SDK call;
+    // returns immediately when the worker never started.
+    static void DrainAndWait();
+};
+
+}  // namespace ubring
+}  // namespace brpc
+
+#endif //BRPC_UB_CLEANUP_WORKER_H
diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp
index 1d551e3315..98f845702e 100644
--- a/src/brpc/ubshm/ub_ring.cpp
+++ b/src/brpc/ubshm/ub_ring.cpp
@@ -23,6 +23,7 @@
 #include "bthread/bthread.h"
 #include "butil/logging.h"
 #include "brpc/ubshm/ub_ring.h"
+#include "brpc/ubshm/ub_cleanup_worker.h"
 #include "brpc/ubshm/ub_ring_manager.h"
 #include "brpc/ubshm/shm/shm_ipc.h"
 
@@ -159,26 +160,32 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, uint64_t expect_ubr_id,
             expect_ubr_id);
     if (BAIDU_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.
+        // timer/callback and starter references. Hand the cleanup to the
+        // worker -- the caller may be the timer thread -- so the trx does not
+        // end up with neither timers nor a queued cleanup, and run it inline
+        // only when the worker cannot be started.
         int state_expected = UBR_CLEANUP_PENDING;
         if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, state_expected, UBR_CLEANUP_RUNNING)) {
-            // Gate on the slot still being used by this generation: force
-            // close may have claimed the cleanup and released the slot (which
-            // already freed the trx resources) after we were anchored.
-            if (UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id)) {
-                work(trx, ctl->ubr_id);
+            if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id, work, ctl)) {
+                // Gate on the slot still being used by this generation: force
+                // close may have claimed the cleanup and released the slot
+                // (which already freed the trx resources) after we were
+                // anchored.
+                if (UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id)) {
+                    work(trx, ctl->ubr_id);
+                }
+                ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
+                ctl->ReleaseRef();           // timer/callback reference
             }
-            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.
-            UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl);
+            // When the job was posted, the worker owns the timer/callback
+            // reference and settles state/anchor exactly like the timer-fired
+            // path; the manager anchor keeps the control object alive until
+            // the slot is acquired again or UbrMgrFini retires it.
+        } else {
+            // A force close already owns the cleanup. It could not take the
+            // reference through the never-armed timer, so release it here.
+            ctl->ReleaseRef();               // timer/callback reference
         }
-        ctl->ReleaseRef();                   // timer/callback reference
         ctl->ReleaseRef();                   // starter reference
         return UBRING_ERR;
     }
@@ -241,13 +248,16 @@ RETURN_CODE UBRing::UbrTrxClose() {
             // to an already running delayed-clear callback otherwise.
             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
-            }
+            // Snapshot the delayed cleanup and, when there is none, claim the
+            // cleanup for this force close in one critical section. Splitting
+            // the two would let a concurrent SDK-fault callback publish a new
+            // cleanup after our empty snapshot, leaving two paths running the
+            // cleanup of the same shared memory.
+            UbrCleanupCtl* ctl = nullptr;
+            const UbrCleanupClaim claim = UBRingManager::ClaimTrxCleanupForced(
+                    _trx->trx_mgr_index, expect_ubr_id, &ctl);
             bool cleanup_owned = false;
-            if (ctl != nullptr) {
+            if (claim == UBR_CLEANUP_CLAIM_HAS_CTL) {
                 int expected = UBR_CLEANUP_PENDING;
                 if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, expected, UBR_CLEANUP_RUNNING)) {
                     cleanup_owned = true;
@@ -255,7 +265,7 @@ RETURN_CODE UBRing::UbrTrxClose() {
                         ctl->ReleaseRef();   // timer/callback reference
                     }
                 }
-            } else if (ATOMIC_LOAD(_trx->ubr_id) == expect_ubr_id) {
+            } else if (claim == UBR_CLEANUP_CLAIM_OWNED_NULL) {
                 cleanup_owned = true;
             }
             if (cleanup_owned) {
@@ -408,10 +418,16 @@ RETURN_CODE UBRing::UbrAddTimer() {
 // A slot cannot be released and reused underneath a callback that already
 // passed that check, and the guarantee is a synchronization one, not a timing
 // one. bthread dispatches all timer callbacks from a single global timer
-// thread, so two per-trx callbacks never run concurrently, and a cleanup that
-// runs from a one-shot timer callback (the delayed-clear path) is serialized
-// after every per-trx callback that was dispatched before it. Entering a close
-// claims the slot for the generation under g_ubr_trx_mgr_mtx
+// thread, so two per-trx callbacks never run concurrently, and the delayed
+// clear arbitration (the ctl->state compare-exchange in the clear callbacks)
+// is serialized after every per-trx callback that was dispatched before it.
+// The cleanup body itself no longer runs on the timer thread but on the
+// cleanup worker (UbrCleanupWorker): it is kept away from a released or reused
+// slot by the generation gate it re-applies under g_ubr_trx_mgr_mtx right
+// before running the work (IsUbrTrxSlotUsed), and by the scheduling paths that
+// wait for the per-trx callbacks through UbrStopTrxTimer -> UbrTimerDelAndWait
+// before they may clear the trx or hand its slot on. Entering a close claims
+// the slot for the generation under g_ubr_trx_mgr_mtx
 // (UBRingManager::TryClaimTrxClose), so a caller working from a stale snapshot
 // -- the faulty-shm event, which looks the slot up before the claim -- cannot
 // touch the next occupant. Teardown that happens on any other thread then waits
@@ -538,12 +554,19 @@ void* UBRing::UbrPassiveClearCallback(void* args, uint64_t) {
         ctl->ReleaseRef();
         return nullptr;
     }
+    // The ownership arbitration above is settled; hand the blocking work to
+    // the cleanup worker instead of running it on the timer thread, which
+    // serves every timer in the process. The job takes over the timer/callback
+    // reference.
     UbrTrx* trx = ctl->trx;
-    if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) {
-        UbrDoPassiveClearWork(trx, ctl->ubr_id);
+    if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id,
+                                          UbrDoPassiveClearWork, ctl)) {
+        if (BAIDU_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
     }
-    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
-    ctl->ReleaseRef();                       // timer/callback reference
     return nullptr;
 }
 
@@ -615,12 +638,17 @@ void *UBRing::UbrAsynClearCallback(void *args, uint64_t)
         ctl->ReleaseRef();
         return nullptr;
     }
+    // Same offload as the passive path: the timer thread only arbitrates
+    // ownership, the worker does the SDK-backed cleanup.
     UbrTrx* trx = ctl->trx;
-    if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) {
-        UbrDoAsynClearWork(trx, ctl->ubr_id);
+    if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id,
+                                          UbrDoAsynClearWork, ctl)) {
+        if (BAIDU_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
     }
-    ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE);
-    ctl->ReleaseRef();                       // timer/callback reference
     return nullptr;
 }
 
diff --git a/src/brpc/ubshm/ub_ring_manager.cpp b/src/brpc/ubshm/ub_ring_manager.cpp
index 89529e37e6..c5c654f98b 100644
--- a/src/brpc/ubshm/ub_ring_manager.cpp
+++ b/src/brpc/ubshm/ub_ring_manager.cpp
@@ -251,6 +251,7 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) {
             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;
+            g_ubr_mgr.trx_mgr[i].cleanup_forced.store(false);
             // 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;
@@ -329,16 +330,45 @@ RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx,
     return UBRING_OK;
 }
 
-UbrCleanupCtl* UBRingManager::SnapshotUnitCleanupCtl(uint32_t idx) {
+UbrCleanupClaim UBRingManager::ClaimTrxCleanupForced(
+        uint32_t idx, uint64_t expect_ubr_id, UbrCleanupCtl** out_ctl) {
+    if (out_ctl != nullptr) {
+        *out_ctl = nullptr;
+    }
     BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx);
-    if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || idx >= g_ubr_mgr.trx_cap)) {
-        return nullptr;
+    if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr == 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 ||
+                 idx >= g_ubr_mgr.trx_cap)) {
+        return UBR_CLEANUP_CLAIM_NOT_OURS;
+    }
+    // Same generation check as TryClaimTrxClose: a caller that snapshotted the
+    // slot before a release/reuse must not claim the new occupant's cleanup.
+    if (g_ubr_mgr.trx_mgr_unit_status[idx] != UBR_MGR_UNIT_USED ||
+        g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id) {
+        return UBR_CLEANUP_CLAIM_NOT_OURS;
     }
     UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[idx];
     if (ctl != nullptr) {
+        // An anchored ctl always belongs to the current generation: the
+        // publication checked it and the acquire retires the previous one.
+        // Keep the explicit check as a belt-and-braces guard.
+        if (BAIDU_UNLIKELY(ctl->ubr_id != expect_ubr_id)) {
+            return UBR_CLEANUP_CLAIM_NOT_OURS;
+        }
         ctl->ref.fetch_add(1);               // snapshot reference
-    }
-    return ctl;
+        if (out_ctl != nullptr) {
+            *out_ctl = ctl;
+        }
+        return UBR_CLEANUP_CLAIM_HAS_CTL;
+    }
+    // No cleanup was published: this call owns the forced cleanup, and
+    // refusing later publications keeps it the only owner. TryPublishUnitCleanupCtl
+    // checks the flag in this same critical section, so a publication either
+    // happened before (we would have seen its ctl above) or is refused.
+    g_ubr_mgr.trx_mgr[idx].cleanup_forced.store(true);
+    return UBR_CLEANUP_CLAIM_OWNED_NULL;
 }
 
 bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) {
@@ -425,8 +455,9 @@ bool UBRingManager::TryPublishUnitCleanupCtl(uint32_t idx,
                  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[idx].cleanup_forced.load() ||
                  g_ubr_mgr.trx_mgr_unit_ctl[idx] != nullptr)) {
-        return false;                        // shutting down / released / reused / already anchored
+        return false;                        // shutting down / released / reused / forced / already anchored
     }
     // Publish on the trx and anchor it in the pool slot inside one critical
     // section: a force close that snapshots the slot (also under this lock)
diff --git a/src/brpc/ubshm/ub_ring_manager.h b/src/brpc/ubshm/ub_ring_manager.h
index 298bbf218e..711c06e339 100644
--- a/src/brpc/ubshm/ub_ring_manager.h
+++ b/src/brpc/ubshm/ub_ring_manager.h
@@ -31,6 +31,19 @@ typedef enum {
     UBR_MGR_UNIT_USED = 1
 } UbrMgrUnitStatus;
 
+// Result of UBRingManager::ClaimTrxCleanupForced.
+typedef enum {
+    // The slot is not (or no longer) used by the requested generation: the
+    // caller must not touch it and must not run the cleanup.
+    UBR_CLEANUP_CLAIM_NOT_OURS = 0,
+    // A delayed cleanup is published for that generation and was handed back
+    // in *out_ctl with a snapshot reference the caller must release.
+    UBR_CLEANUP_CLAIM_HAS_CTL = 1,
+    // No delayed cleanup existed; the forced claim took the cleanup ownership
+    // and later publications for that generation are refused.
+    UBR_CLEANUP_CLAIM_OWNED_NULL = 2
+} UbrCleanupClaim;
+
 typedef struct TagUbrMgr {
     uint32_t trx_num;
     uint32_t trx_cap;
@@ -95,9 +108,23 @@ class UBRingManager {
     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);
+    // Snapshot the delayed cleanup control object of a pool slot and, when
+    // there is none, claim the cleanup of `expect_ubr_id' for the force close
+    // that will run it. Both the snapshot and the claim happen under the
+    // manager lock, i.e. in the same critical section as
+    // TryPublishUnitCleanupCtl, so the two are totally ordered: either the
+    // publication ran first (UBR_CLEANUP_CLAIM_HAS_CTL, and the existing
+    // ctl->state arbitration decides who runs the cleanup) or the claim ran
+    // first (UBR_CLEANUP_CLAIM_OWNED_NULL, and every later publication for
+    // this generation is refused through trx->cleanup_forced). Splitting the
+    // two steps -- a snapshot, then an out-of-lock re-check that no ctl
+    // appeared -- leaves a window in which a concurrent SDK-fault callback
+    // publishes a new cleanup after an empty snapshot and both paths run the
+    // cleanup. `*out_ctl' is only written on UBR_CLEANUP_CLAIM_HAS_CTL, and
+    // carries a snapshot reference the caller must ReleaseRef.
+    static UbrCleanupClaim ClaimTrxCleanupForced(uint32_t idx,
+                                                 uint64_t expect_ubr_id,
+                                                 UbrCleanupCtl** out_ctl);
 
     // Under the manager lock, confirm the pool slot is still used by the
     // given generation (not released or reused meanwhile).
@@ -121,7 +148,8 @@ class UBRingManager {
     // other. 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.
+    // released, reused, a force close already claimed the cleanup, or a
+    // cleanup is already anchored.
     static bool TryPublishUnitCleanupCtl(uint32_t idx, uint64_t expect_ubr_id,
                                          UbrCleanupCtl *ctl);
 
diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h
index cc35f7223f..5a489117a7 100644
--- a/src/brpc/ubshm/ubr_trx.h
+++ b/src/brpc/ubshm/ubr_trx.h
@@ -170,6 +170,15 @@ typedef struct TagUbrTrx {
     butil::atomic<UbrTimerId> close_timer;
     butil::atomic<UbrTimerId> hb_timer;
     butil::atomic<UbrCleanupCtl*> cleanup_ctl;
+    // True once a force close claimed the cleanup of this acquisition while no
+    // delayed cleanup was published (see
+    // UBRingManager::ClaimTrxCleanupForced). The forced claim and the
+    // publication of a delayed cleanup are serialized by the manager lock, so
+    // a later publication sees the flag and refuses: only one of the two ever
+    // owns the cleanup. Reset by AcquireUbrTrxFromMgr together with the other
+    // per-acquisition fields; a stale true on a FREE slot is harmless because
+    // a publication is already refused while the slot is not USED.
+    butil::atomic<bool> cleanup_forced;
     // 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;