fix: prevent dirty_data_key_count_ underflow from three sources

tx_service/include/cc/cc_request.h

@@ -6229,8 +6229,61 @@ struct KickoutCcEntryCc : public TemplatedCcRequest<KickoutCcEntryCc, Void>
         case CleanType::CleanRangeData:
         case CleanType::CleanRangeDataForMigration:
         case CleanType::CleanBucketData:
-            // All data in the target range/bucket can be cleaned.
-            return true;
+        {
+            // Cannot clean entries being checkpointed — the checkpoint
+            // callback (UpdateCceCkptTsCc) will decrement dirty count,
+            // and cleaning here would cause a double-decrement.
+            //
+            // An alternative approach would be to have
+            // DataSyncForHashPartition acquire bucket read locks (as
+            // DataSyncForRangePartition already does), which would
+            // serialize checkpoint with migration and prevent this race
+            // entirely. However, that has significant downsides for hash
+            // partitions: checkpoint is dispatched per-core, and each
+            // core owns 1024/num_cores buckets (e.g. 128 on an 8-core
+            // node), so locking all of them via ReadTxRequest adds
+            // overhead to every checkpoint cycle even when no migration
+            // is in progress.
+            //
+            // Instead, we skip entries with BeingCkpt=true here and
+            // rely on the KickoutCcEntryCc retry loop
+            // (template_cc_map.h, Execute(KickoutCcEntryCc&)) to
+            // re-enqueue the request. Once the checkpoint callback
+            // clears BeingCkpt, the entry is cleaned on the next retry.
+            // This only delays migration when there is an active
+            // checkpoint on the same bucket — a rare overlap in
+            // practice.
+            if (versioned_cce)
+            {
+                if (range_partitioned)
+                {
+                    return !static_cast<const VersionedLruEntry<true, true> *>(
+                                entry)
+                                ->GetBeingCkpt();
+                }
+                else
+                {
+                    return !static_cast<const VersionedLruEntry<true, false> *>(
+                                entry)
+                                ->GetBeingCkpt();
+                }
+            }
+            else
+            {
+                if (range_partitioned)
+                {
+                    return !static_cast<const VersionedLruEntry<false, true> *>(
+                                entry)
+                                ->GetBeingCkpt();
+                }
+                else
+                {
+                    return !static_cast<
+                                const VersionedLruEntry<false, false> *>(entry)
+                                ->GetBeingCkpt();
+                }
+            }
+        }
         case CleanType::CleanForAlterTable:
         {
             if (versioned_cce)

tx_service/include/cc/cluster_config_cc_map.h

@@ -272,8 +272,10 @@ class ClusterConfigCcMap
 
         if (lk_type != LockType::NoLock)
         {
+            bool was_dirty = neg_inf_.IsDirty();
             neg_inf_.SetCommitTsPayloadStatus(req.CommitTs(),
                                               RecordStatus::Normal);
+            OnCommittedUpdate(&neg_inf_, was_dirty);
             ReleaseCceLock(lock, &neg_inf_, txn, req.NodeGroupId());
         }
 
@@ -526,8 +528,13 @@ class ClusterConfigCcMap
                 cluster_scale_txn, tx_candidate_term, req.CommitTs());
             txm->RecoverClusterScale(scale_op_msg, dm_started, dm_finished);
         }
-        neg_inf_.SetCommitTsPayloadStatus(
-            Sharder::Instance().ClusterConfigVersion(), RecordStatus::Normal);
+        {
+            bool was_dirty = neg_inf_.IsDirty();
+            neg_inf_.SetCommitTsPayloadStatus(
+                Sharder::Instance().ClusterConfigVersion(),
+                RecordStatus::Normal);
+            OnCommittedUpdate(&neg_inf_, was_dirty);
+        }
 
         req.SetFinish();
         return true;

tx_service/include/cc/template_cc_map.h

@@ -2290,6 +2290,7 @@ class TemplateCcMap : public CcMap
         // backfill version.
         cce->SetCkptTs(req.CommitTs());
         OnFlushed(cce, was_dirty);
+        OnCommittedUpdate(cce, was_dirty);
 
         // Refill mvcc archives.
         if (shard_->EnableMvcc())
@@ -10119,8 +10120,6 @@ class TemplateCcMap : public CcMap
         const uint64_t cce_version = cce->CommitTs();
 
         bool was_dirty = cce->IsDirty();
-        cce->SetCkptTs(commit_ts);
-        OnFlushed(cce, was_dirty);
 
         if (cce_version < commit_ts)
         {
@@ -10139,6 +10138,7 @@ class TemplateCcMap : public CcMap
                     return false;
                 }
             }
+            // Update commit ts and payload status first (clears flush bit).
             cce->SetCommitTsPayloadStatus(commit_ts, status);
 
             if (status == RecordStatus::Deleted)
@@ -10163,6 +10163,11 @@ class TemplateCcMap : public CcMap
             payload->Deserialize(rec_str.c_str(), offset);
             cce->AddArchiveRecord(payload, status, commit_ts);
         }
+        // Set ckpt ts after SetCommitTsPayloadStatus so that the flush
+        // bit is not cleared by the overwrite.
+        cce->SetCkptTs(commit_ts);
+        OnFlushed(cce, was_dirty);
+        OnCommittedUpdate(cce, was_dirty);
         if (RangePartitioned && cce->entry_info_.DataStoreSize() == INT32_MAX)
         {
             cce->entry_info_.SetDataStoreSize(

tx_service/src/cc/cc_req_misc.cpp

@@ -1164,7 +1164,7 @@ bool UpdateCceCkptTsCc::Execute(CcShard &ccs)
                 VersionedLruEntry<true, true> *v_entry =
                     static_cast<VersionedLruEntry<true, true> *>(ref.cce_);
 
-                assert(v_entry->CommitTs() > 1 && !v_entry->IsPersistent());
+                assert(v_entry->CommitTs() > 1);
                 bool was_dirty = v_entry->IsDirty();
                 v_entry->entry_info_.SetDataStoreSize(ref.post_flush_size_);
 
@@ -1176,7 +1176,7 @@ bool UpdateCceCkptTsCc::Execute(CcShard &ccs)
             {
                 VersionedLruEntry<false, true> *v_entry =
                     static_cast<VersionedLruEntry<false, true> *>(ref.cce_);
-                assert(v_entry->CommitTs() > 1 && !v_entry->IsPersistent());
+                assert(v_entry->CommitTs() > 1);
                 bool was_dirty = v_entry->IsDirty();
                 v_entry->entry_info_.SetDataStoreSize(ref.post_flush_size_);
 
@@ -1192,7 +1192,7 @@ bool UpdateCceCkptTsCc::Execute(CcShard &ccs)
                 VersionedLruEntry<true, false> *v_entry =
                     static_cast<VersionedLruEntry<true, false> *>(ref.cce_);
 
-                assert(v_entry->CommitTs() > 1 && !v_entry->IsPersistent());
+                assert(v_entry->CommitTs() > 1);
                 bool was_dirty = v_entry->IsDirty();
                 v_entry->SetCkptTs(ref.commit_ts_);
                 v_entry->ClearBeingCkpt();
@@ -1203,7 +1203,7 @@ bool UpdateCceCkptTsCc::Execute(CcShard &ccs)
                 VersionedLruEntry<false, false> *v_entry =
                     static_cast<VersionedLruEntry<false, false> *>(ref.cce_);
 
-                assert(v_entry->CommitTs() > 1 && !v_entry->IsPersistent());
+                assert(v_entry->CommitTs() > 1);
                 bool was_dirty = v_entry->IsDirty();
                 v_entry->SetCkptTs(ref.commit_ts_);
                 v_entry->ClearBeingCkpt();

tx_service/src/cc/cc_shard.cpp

@@ -445,12 +445,8 @@ void CcShard::AdjustDataKeyStats(const TableName &table_name,
 
     if (dirty_delta != 0)
     {
-        // Sanity check in debug mode.
         assert(dirty_delta >= 0 ||
                dirty_data_key_count_ >= static_cast<size_t>(-dirty_delta));
-        // Clamp to avoid underflow when an entry is flushed but was never
-        // counted dirty (e.g. became dirty via a path that doesn't call
-        // OnCommittedUpdate).
         int64_t new_dirty =
             static_cast<int64_t>(dirty_data_key_count_) + dirty_delta;
         if (new_dirty < 0)