diff --git a/tx_service/include/cc/cc_request.h b/tx_service/include/cc/cc_request.h index 47ebfce8..b5ebfb73 100644 --- a/tx_service/include/cc/cc_request.h +++ b/tx_service/include/cc/cc_request.h @@ -6229,8 +6229,61 @@ struct KickoutCcEntryCc : public TemplatedCcRequest 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 *>( + entry) + ->GetBeingCkpt(); + } + else + { + return !static_cast *>( + entry) + ->GetBeingCkpt(); + } + } + else + { + if (range_partitioned) + { + return !static_cast *>( + entry) + ->GetBeingCkpt(); + } + else + { + return !static_cast< + const VersionedLruEntry *>(entry) + ->GetBeingCkpt(); + } + } + } case CleanType::CleanForAlterTable: { if (versioned_cce) diff --git a/tx_service/include/cc/cluster_config_cc_map.h b/tx_service/include/cc/cluster_config_cc_map.h index 3e55acb9..f9d7c34d 100644 --- a/tx_service/include/cc/cluster_config_cc_map.h +++ b/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; diff --git a/tx_service/include/cc/template_cc_map.h b/tx_service/include/cc/template_cc_map.h index 1870cbe5..17cd2327 100644 --- a/tx_service/include/cc/template_cc_map.h +++ b/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( diff --git a/tx_service/src/cc/cc_req_misc.cpp b/tx_service/src/cc/cc_req_misc.cpp index 2d6dbf31..f672a23b 100644 --- a/tx_service/src/cc/cc_req_misc.cpp +++ b/tx_service/src/cc/cc_req_misc.cpp @@ -1164,7 +1164,7 @@ bool UpdateCceCkptTsCc::Execute(CcShard &ccs) VersionedLruEntry *v_entry = static_cast *>(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 *v_entry = static_cast *>(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 *v_entry = static_cast *>(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 *v_entry = static_cast *>(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(); diff --git a/tx_service/src/cc/cc_shard.cpp b/tx_service/src/cc/cc_shard.cpp index 661ceb5f..5a3ba7bc 100644 --- a/tx_service/src/cc/cc_shard.cpp +++ b/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(-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(dirty_data_key_count_) + dirty_delta; if (new_dirty < 0)