fix(kernel): retire deferred updates with destroyed entities

docs/reference/kernel.md

@@ -304,9 +304,11 @@ Changes with `sync_view` or `sync_self` and without `real_time` are collected un
 
 Each `class_call_back_manager::on_delay_sync()` call detaches its current collection before invoking callbacks. Entries collected during a callback remain pending for the next flush instead of being discarded by an end-of-flush clear. A recursive flush returns `false` without consuming pending changes. If a callback throws, the exception propagates, the flush guard resets and newly collected entries remain pending; the detached batch is not automatically replayed.
 
+Kernel entity destruction completes its class destruction callbacks, then cancels that entity's queued delayed updates before removing it from the registry. Retirement also removes a later entity from an active batch or skips the current entity's remaining masks when a delivery callback destroys it. Live entities, class destruction events and surviving parent-container destruction markers remain intact. Recreating the same GUID cannot replay the retired lifetime's batch; new changes wait for the next flush.
+
 `kernel_module_impl::update()` retains its existing behavior of ignoring the collector's boolean result and returning `true` when no exception escapes. A nested Kernel update therefore does not report the recursive-flush rejection to the application.
 
-This boundary fixes collection membership, not a point-in-time snapshot of values. Delayed scalar and table entries still borrow `node*` pointers, and serialization reads their current values. If one mask callback changes a scalar already in the detached batch, a later mask can read that new value in the same flush; the newly collected entry remains pending for the next flush as well. Keep referenced nodes and their owning entities/rows alive until all relevant pending and active batches finish. Do not destroy or replace them from a delivery callback. The collector remains single-threaded and does not provide ownership, retry, rollback or delivery acknowledgement.
+This boundary fixes collection membership, not a point-in-time snapshot of values. Delayed scalar and table entries still borrow `node*` pointers, and serialization reads their current values. If one mask callback changes a scalar already in the detached batch, a later mask can read that new value in the same flush; the newly collected entry remains pending for the next flush as well. Kernel retirement cancels the entity's batches; otherwise keep referenced nodes and rows alive until all relevant pending and active batches finish. Do not replace rows or mutate retained retired entities from a delivery callback. The collector remains single-threaded and does not provide ownership, retry, rollback or delivery acknowledgement.
 
 `check-kernel-initialization` exercises real container mutations through delayed collection, Kernel update, serialization and the existing delivery callback, with empty, scalar, table and mixed controls. Its test-only callback observer does not provide network delivery. The existing `send_to_self()` path remains a stub; these checks do not establish a client replication protocol, acknowledgements or AOI behavior.
 

docs/reference/kernel_CN.md

@@ -298,9 +298,11 @@ private:
 
 每次 `class_call_back_manager::on_delay_sync()` 在调用回调前分离当前收集集合。回调期间收集的新条目保留到下一次刷新，不会被刷新末尾的清空操作丢弃。递归刷新返回 `false`，不消耗待刷新变更。回调抛出异常时，异常继续向外传播，刷新防护复位，新收集的条目仍保持待刷新状态；分离出的批次不会自动重放。
 
+Kernel 销毁实体时先完成其类销毁回调，再取消该实体的待刷新延迟更新，最后从注册表移除对象。交付回调内销毁实体时，退休操作也会移除当前批次中尚未分发的该实体，或跳过正在分发实体的剩余掩码。存活实体、类销毁事件及仍存活父容器的销毁标记保持不变。同一 GUID 重建后不会重放旧生命周期的批次；新变更等到下一次刷新。
+
 `kernel_module_impl::update()` 保留原有行为：忽略收集器的布尔返回值，在没有异常向外传播时返回 `true`。因此，嵌套调用 Kernel 更新不会向应用报告递归刷新被拒绝。
 
-此边界固定批次条目，不提供某一时刻的值快照。延迟标量和表格条目仍借用 `node*` 指针，序列化读取当时的值。如果前一个掩码回调修改了已在分离批次中的标量，后一个掩码仍可能在同次刷新中读到新值；新收集的条目也会保留到下一次刷新。在所有相关待刷新批次和当前批次处理完之前，必须保持引用的节点及其所属实体／行存活，不要在交付回调中销毁或替换它们。收集器仍按单线程使用，不提供所有权、重试、回滚或交付确认。
+此边界固定批次条目，不提供某一时刻的值快照。延迟标量和表格条目仍借用 `node*` 指针，序列化读取当时的值。如果前一个掩码回调修改了已在分离批次中的标量，后一个掩码仍可能在同次刷新中读到新值；新收集的条目也会保留到下一次刷新。Kernel 退休操作会取消该实体的批次；其余情况下，在所有相关待刷新批次和当前批次处理完之前，必须保持引用的节点及行存活。不要在交付回调中替换行，或修改已退休但仍持有引用的实体。收集器仍按单线程使用，不提供所有权、重试、回滚或交付确认。
 
 `check-kernel-initialization` 从真实容器操作经过延迟收集、Kernel 更新、序列化和既有交付回调验证此路径，并保留空批次、标量、表格和混合对照。仅用于测试的回调观察器不提供网络交付。现有 `send_to_self()` 仍是占位实现；这些检查不证明客户端复制协议、确认机制或 AOI 行为。
 

src/plugins/kernel/class_call_back_manager.cpp

@@ -22,6 +22,19 @@
 
 namespace ark {
 
+namespace {
+
+struct delayed_sync_dispatch
+{
+    class_call_back_manager::delay_sync_data_list pending;
+    const guid_t* current{};
+    bool retired{};
+};
+
+delayed_sync_dispatch* active_delay_sync{};
+
+} // namespace
+
 class_call_back_manager::node_sync_call_backs class_call_back_manager::node_sync_call_back_list_;
 class_call_back_manager::table_sync_call_backs class_call_back_manager::table_sync_call_back_list_;
 class_call_back_manager::container_sync_call_back class_call_back_manager::container_sync_call_backs_list_;
@@ -262,34 +275,38 @@ void class_call_back_manager::add_delay_sync_call_back(const data_mask mask_valu
 }
 
 bool class_call_back_manager::on_delay_sync() {
-    static bool flushing = false;
-    if (flushing)
+    if (active_delay_sync)
     {
         return false;
     }
 
     struct flush_guard
     {
-        bool& active;
-
         ~flush_guard() {
-            active = false;
+            active_delay_sync = nullptr;
         }
     };
 
-    flushing = true;
-    flush_guard guard{flushing};
-
     // Newly collected entries remain pending for the next flush, even on throw.
     // Borrowed node values are not frozen. The detached batch is not replayed.
-    delay_sync_data_list batch;
-    batch.swap(delay_sync_data_list_);
-    for (auto& iter : batch)
-    {
-        auto& guid = iter.first;
-        auto& mask_data_map = iter.second;
-        for (auto& iter_data : mask_data_map)
+    delayed_sync_dispatch dispatch;
+    dispatch.pending.swap(delay_sync_data_list_);
+    active_delay_sync = &dispatch;
+    flush_guard guard;
+    while (!dispatch.pending.empty())
+    {
+        // Extract the current entity before invoking user code. Retirement can
+        // then erase any later entity without invalidating our traversal.
+        auto entry = dispatch.pending.extract(dispatch.pending.begin());
+        auto& guid = entry.key();
+        dispatch.current = &guid;
+        dispatch.retired = false;
+        for (auto& iter_data : entry.mapped())
         {
+            if (dispatch.retired)
+            {
+                break;
+            }
             auto mask_value = iter_data.first;
             auto iter_call_back = delay_sync_call_back_list_.find(mask_value);
             if (iter_call_back == delay_sync_call_back_list_.end())
@@ -299,11 +316,24 @@ bool class_call_back_manager::on_delay_sync() {
 
             iter_call_back->second(guid, mask_value, iter_data.second);
         }
+        dispatch.current = nullptr;
     }
 
     return true;
 }
 
+void class_call_back_manager::retire_delay_sync(const guid_t& self) {
+    delay_sync_data_list_.erase(self);
+    if (active_delay_sync)
+    {
+        active_delay_sync->pending.erase(self);
+        if (active_delay_sync->current && *active_delay_sync->current == self)
+        {
+            active_delay_sync->retired = true;
+        }
+    }
+}
+
 class_call_back_manager::delay_sync_mask_data& class_call_back_manager::get_delay_sync_mask_data(const guid_t& self) {
     auto iter_data = delay_sync_data_list_.find(self);
     if (iter_data == delay_sync_data_list_.end())

src/plugins/kernel/class_call_back_manager.hpp

@@ -160,6 +160,9 @@ class class_call_back_manager final
     static bool on_delay_sync();
 
 private:
+    friend class kernel_module_impl;
+    static void retire_delay_sync(const guid_t& self);
+
     delay_sync_mask_data& get_delay_sync_mask_data(const guid_t& self);
     delay_sync_table& get_delay_sync_mask_table(
         delay_sync_mask_data& mask_data_map, const data_mask mask_value, uint32_t index);

src/plugins/kernel/kernel_module_impl.cpp

@@ -379,6 +379,10 @@ bool kernel_module_impl::inner_destroy_entity(std::shared_ptr<entity> entity_ptr
         do_event(self, class_name, entity_event::entity_evt_pre_destroy, data_list_impl());
         do_event(self, class_name, entity_event::entity_evt_destroy, data_list_impl());
 
+        // Retire updates collected before or during destruction callbacks,
+        // including a detached flush. Parent-container destruction markers are
+        // queued under the surviving parent and remain available for delivery.
+        class_call_back_manager::retire_delay_sync(self);
         return objects_.erase(self);
     }
     else

tests/core/kernel_delayed_sync.cpp

@@ -93,7 +93,12 @@ class delayed_sync_world
 public:
     delayed_sync_world() {
         ark::logger::init(&logger_);
-        logger_.set_log_func([](spdlog::level::level_enum, const std::string&) {});
+        logger_.set_log_func([this](spdlog::level::level_enum level, const std::string& message) {
+            if (level >= spdlog::level::err)
+            {
+                errors.push_back(message);
+            }
+        });
         add<ark::class_meta_module>(metadata_);
         add<ark::config_module>(config_);
         add<ark::kernel_module>(kernel);
@@ -168,6 +173,7 @@ class delayed_sync_world
     ark::kernel_module_impl kernel;
     std::shared_ptr<ark::entity> owner;
     std::vector<delivery> deliveries;
+    std::vector<std::string> errors;
     std::function<void(ark::data_mask)> on_delivery;
 
 private:
@@ -189,6 +195,131 @@ class delayed_sync_world
 
 } // namespace
 
+TEST_CASE("Kernel retirement cancels queued updates while preserving live entities and destroy events") {
+    struct destroy_observer
+    {
+        std::vector<ark::entity_event> events;
+        std::shared_ptr<ark::entity> owner;
+
+        int on_event(const ark::guid_t& id, const std::string&, ark::entity_event event, const ark::data_list&) {
+            if (id == 10 &&
+                (event == ark::entity_event::entity_evt_pre_destroy || event == ark::entity_event::entity_evt_destroy))
+            {
+                events.push_back(event);
+                // Destruction callbacks may collect one final old-object change.
+                REQUIRE(owner->set_u_int32(100, 80 + static_cast<std::uint32_t>(events.size())));
+            }
+            return 0;
+        }
+    } observed;
+
+    delayed_sync_world world;
+    observed.owner = world.owner;
+    REQUIRE(static_cast<ark::kernel_module&>(world.kernel)
+                .add_class_call_back("Owner", &observed, &destroy_observer::on_event, 100));
+    const auto live = world.kernel.create_entity(20, 1, 0, "Owner", 0, ark::data_list_impl{});
+    REQUIRE(live != nullptr);
+    REQUIRE(world.owner->set_u_int32(100, 7));
+    REQUIRE(live->set_u_int32(100, 29));
+    const std::weak_ptr<ark::entity> retired = world.owner;
+    REQUIRE(world.kernel.destroy_entity(10));
+    world.owner.reset();
+    observed.owner.reset();
+    CHECK(retired.expired());
+    CHECK(observed.events == std::vector<ark::entity_event>{
+                                 ark::entity_event::entity_evt_pre_destroy, ark::entity_event::entity_evt_destroy});
+    REQUIRE(world.kernel.update());
+    CHECK(world.errors.empty());
+    REQUIRE(world.deliveries.size() == 2);
+    for (const auto& event : world.deliveries)
+    {
+        CHECK(event.target == 20);
+        CHECK(event.message.data().datas_uint32().at(100) == 29);
+    }
+}
+
+TEST_CASE("Kernel retirement during delivery skips later masks of the active entity lifetime") {
+    delayed_sync_world world;
+    REQUIRE(world.owner->set_u_int32(100, 7));
+    const auto old_owner = world.owner;
+    bool retired = false;
+    bool recreate = false;
+    SUBCASE("ordinary retirement") {
+    }
+    SUBCASE("same GUID creates a distinct lifetime for the next flush") {
+        recreate = true;
+    }
+    world.on_delivery = [&](ark::data_mask) {
+        if (retired)
+        {
+            return;
+        }
+        retired = true;
+        REQUIRE(world.kernel.destroy_entity(10));
+        if (recreate)
+        {
+            world.owner = world.kernel.create_entity(10, 1, 0, "Owner", 0, ark::data_list_impl{});
+            REQUIRE(world.owner != nullptr);
+            REQUIRE(world.owner != old_owner);
+            REQUIRE(world.owner->set_u_int32(100, 99));
+        }
+    };
+    REQUIRE(world.kernel.update());
+    CHECK(retired);
+    CHECK(world.errors.empty());
+    REQUIRE(world.deliveries.size() == 1);
+    CHECK(world.deliveries.front().mask == ark::data_mask::pf_sync_view);
+    CHECK(world.deliveries.front().message.data().datas_uint32().at(100) == 7);
+    world.on_delivery = {};
+    world.deliveries.clear();
+    if (recreate)
+    {
+        world.flush();
+        for (const auto& event : world.deliveries)
+        {
+            CHECK(event.message.data().datas_uint32().at(100) == 99);
+        }
+    }
+    else
+    {
+        REQUIRE(world.kernel.update());
+        CHECK(world.deliveries.empty());
+    }
+    CHECK(world.errors.empty());
+}
+
+TEST_CASE("Kernel retirement during delivery removes another queued lifetime without losing its neighbors") {
+    delayed_sync_world world;
+    auto victim = world.kernel.create_entity(20, 1, 0, "Owner", 0, ark::data_list_impl{});
+    const auto live = world.kernel.create_entity(30, 1, 0, "Owner", 0, ark::data_list_impl{});
+    REQUIRE(victim != nullptr);
+    REQUIRE(live != nullptr);
+    REQUIRE(world.owner->set_u_int32(100, 7));
+    REQUIRE(victim->set_u_int32(100, 17));
+    REQUIRE(live->set_u_int32(100, 27));
+    const std::weak_ptr<ark::entity> retired_lifetime = victim;
+    bool retired = false;
+    world.on_delivery = [&](ark::data_mask) {
+        if (!retired)
+        {
+            retired = true;
+            REQUIRE(world.kernel.destroy_entity(20));
+            victim.reset();
+            CHECK(retired_lifetime.expired());
+        }
+    };
+    REQUIRE(world.kernel.update());
+    CHECK(world.errors.empty());
+    REQUIRE(world.deliveries.size() == 4);
+    std::map<ark::guid_t, unsigned> counts;
+    for (const auto& event : world.deliveries)
+    {
+        ++counts[event.target];
+        CHECK(event.message.data().datas_uint32().at(100) == (event.target == 10 ? 7 : 27));
+    }
+    CHECK(counts == std::map<ark::guid_t, unsigned>{{10, 2}, {30, 2}});
+}
+
 TEST_CASE("Kernel delayed container-only changes reach existing mask delivery callbacks") {
     delayed_sync_world world;
     const auto bag = world.owner->find_container(201);