Repository: tikv/tikv
File: components/cdc/src/delegate.rs (Rust)

Code before:
        match &mut self.lock_tracker {
            LockTracker::Pending => unreachable!(),
            LockTracker::Preparing(locks) => {
                let bytes = key.approximate_heap_size();
                self.memory_quota.alloc(bytes)?;
                CDC_PENDING_BYTES_GAUGE.add(bytes as _);
                locks.push(PendingLock::Untrack { key, start_ts });
            }
            LockTracker::Prepared { locks, .. } => {
                if let BTreeMapEntry::Occupied(x) = locks.entry(key.clone()) {
                    if x.get().ts == start_ts {
                        let (key, _) = x.remove_entry();
                        let bytes = key.approximate_heap_size();
                        self.memory_quota.free(bytes);
                        CDC_PENDING_BYTES_GAUGE.sub(bytes as _);
                        lock_count_modify = -1;
                    } else {
                        error!("[for debug] cdc pop_lock found lock with same key but different start_ts";
                            "generation" => ?x.get().generation,
                            "old_start_ts" => ?x.get().ts,
                            "new_start_ts" => ?start_ts,
                            "key" => ?key,
                            "region_id" => self.region_id,
                        );
                        // TODO: remove this in production or new release, ref https://github.com/tikv/tikv/issues/18498.
                        panic!(
                            "[for debug] cdc pop_lock found lock with same key but different start_ts, \
                        generation={:?}, old_start_ts={:?}, new_start_ts={:?}, key={:?}, region_id={:?}",
                            x.get().generation,
                            x.get().ts,
                            start_ts,
                            key,
                            self.region_id,
                        );
                    }
                }
            }
        }
        Ok(lock_count_modify)
    }

    pub(crate) fn init_lock_tracker(&mut self) -> bool {
        if matches!(self.lock_tracker, LockTracker::Pending) {
            self.lock_tracker = LockTracker::Preparing(vec![]);
            return true;
        }
        false
    }

    fn finish_prepare_lock_tracker(
        &mut self,

Review comment (bug):
Is this intentionally not gated behind `ENABLE_DUP_KEY_DEBUG`?

Code after:
        match &mut self.lock_tracker {
            LockTracker::Pending => unreachable!(),
            LockTracker::Preparing(locks) => {
                let bytes = key.approximate_heap_size();
                self.memory_quota.alloc(bytes)?;
                CDC_PENDING_BYTES_GAUGE.add(bytes as _);
                locks.push(PendingLock::Untrack { key, start_ts });
            }
            LockTracker::Prepared { locks, .. } => {
                if let BTreeMapEntry::Occupied(x) = locks.entry(key.clone()) {
                    if x.get().ts == start_ts {
                        let (key, _) = x.remove_entry();
                        let bytes = key.approximate_heap_size();
                        self.memory_quota.free(bytes);
                        CDC_PENDING_BYTES_GAUGE.sub(bytes as _);
                        lock_count_modify = -1;
                    } else {
                        info!("[for debug] cdc pop_lock found lock with same key but different start_ts";
                            "generation" => ?x.get().generation,
                            "old_start_ts" => ?x.get().ts,
                            "new_start_ts" => ?start_ts,
                            "key" => ?key,
                            "region_id" => self.region_id,
                        );
                    }
                }
            }
        }
        Ok(lock_count_modify)
    }

    pub(crate) fn init_lock_tracker(&mut self) -> bool {
        if matches!(self.lock_tracker, LockTracker::Pending) {
            self.lock_tracker = LockTracker::Preparing(vec![]);
            return true;
        }
        false
    }

    fn finish_prepare_lock_tracker(
        &mut self,
        region: Region,
        mut locks: BTreeMap<Key, MiniLock>,
    ) -> Result<()> {
        let delta_locks = match std::mem::replace(&mut self.lock_tracker, LockTracker::Pending) {
            LockTracker::Preparing(locks) => locks,
            _ => unreachable!(),
        };

        let mut free_bytes = 0usize;
        for delta_lock in delta_locks {