Repository: tikv/tikv
File: components/resolved_ts/src/observer.rs (Rust)

Code before:

impl Coprocessor for Observer {}

impl<E: KvEngine> CmdObserver<E> for Observer {
    fn on_flush_applied_cmd_batch(
        &self,
        max_level: ObserveLevel,
        cmd_batches: &mut Vec<CmdBatch>,
        _: &E,
    ) {
        if max_level == ObserveLevel::None {
            return;
        }
        let cmd_batches: Vec<_> = std::mem::take(cmd_batches)
            .into_iter()
            .filter_map(lock_only_filter)
            .collect();
        if cmd_batches.is_empty() {
            return;
        }
        let size = cmd_batches.iter().map(|b| b.size()).sum::<usize>();
        RTS_CHANNEL_PENDING_CMD_BYTES.add(size as i64);
        self.memory_quota.alloc_force(size);
        if let Err(e) = self.scheduler.schedule(Task::ChangeLog {
            cmd_batch: cmd_batches,
        }) {
            info!("failed to schedule change log event"; "err" => ?e);
        }
    }

    fn on_applied_current_term(&self, role: StateRole, region: &Region) {
        // Start to advance resolved ts after peer becomes leader and apply on its term
        if role == StateRole::Leader {
            if let Err(e) = self.scheduler.schedule(Task::RegisterRegion {
                region: region.clone(),
            }) {
                info!("failed to schedule register region task"; "err" => ?e);
            }
        }
    }
}

impl RoleObserver for Observer {
    fn on_role_change(&self, ctx: &mut ObserverContext<'_>, role_change: &RoleChange) {
        // Stop to advance resolved ts after peer steps down to follower or candidate.
        // Do not need to check observe id because we expect all role change events are
        // scheduled in order.
        if role_change.state != StateRole::Leader {
            if let Err(e) = self.scheduler.schedule(Task::DeRegisterRegion {
                region_id: ctx.region().id,
            }) {

Review comment (bug):
Do we need to call `self.memory_quota.free` on the schedule failure path here?

Code after:

impl Coprocessor for Observer {}

impl<E: KvEngine> CmdObserver<E> for Observer {
    fn on_flush_applied_cmd_batch(
        &self,
        max_level: ObserveLevel,
        cmd_batches: &mut Vec<CmdBatch>,
        _: &E,
    ) {
        if max_level == ObserveLevel::None {
            return;
        }
        let cmd_batches: Vec<_> = std::mem::take(cmd_batches)
            .into_iter()
            .filter_map(lock_only_filter)
            .collect();
        if cmd_batches.is_empty() {
            return;
        }
        let size = cmd_batches.iter().map(|b| b.size()).sum::<usize>();
        RTS_CHANNEL_PENDING_CMD_BYTES.add(size as i64);
        self.memory_quota.alloc_force(size);
        if let Err(e) = self.scheduler.schedule(Task::ChangeLog {
            cmd_batch: cmd_batches,
        }) {
            self.memory_quota.free(size);
            info!("failed to schedule change log event"; "err" => ?e);
        }
    }

    fn on_applied_current_term(&self, role: StateRole, region: &Region) {
        // Start to advance resolved ts after peer becomes leader and apply on its term
        if role == StateRole::Leader {
            if let Err(e) = self.scheduler.schedule(Task::RegisterRegion {
                region: region.clone(),
            }) {
                info!("failed to schedule register region task"; "err" => ?e);
            }
        }
    }
}

impl RoleObserver for Observer {
    fn on_role_change(&self, ctx: &mut ObserverContext<'_>, role_change: &RoleChange) {
        // Stop to advance resolved ts after peer steps down to follower or candidate.
        // Do not need to check observe id because we expect all role change events are
        // scheduled in order.
        if role_change.state != StateRole::Leader {
            if let Err(e) = self.scheduler.schedule(Task::DeRegisterRegion {
                region_id: ctx.region().id,