How GreptimeDB's Procedure Framework Manages Distributed Operations Like Region Migration

GreptimeDB's procedure framework provides a fault-tolerant, state-machine-driven runtime that persists execution progress after each step, enabling distributed operations like region migration to survive node crashes and restart exactly where they left off.

The greptimeteam/greptimedb repository implements a robust procedure framework that manages distributed operations with guaranteed durability and isolation. This framework powers region migration, the process of transferring a region's leadership and data between data nodes without service interruption. By combining persistent state machines with exclusive execution guards, the framework ensures that complex multi-step workflows complete reliably even in the face of network failures or node crashes.

Architecture of the Procedure Framework for Distributed Operations

At its core, the framework consists of a global ProcedureManager that coordinates execution, a ProcedureStore that provides durable storage for state snapshots, and a state-machine pattern where each operation implements the Procedure trait. For region migration specifically, the implementation resides in src/meta-srv/src/procedure/region_migration/ and follows a deterministic sequence of states: RegionMigrationStart → OpenCandidateRegion → UpgradeCandidateRegion → RegionMigrationEnd.

The framework operates by repeatedly invoking the State::next method on the current state object. Each invocation returns a Status indicating whether to continue execution, suspend for sub-procedures, finish successfully, or abort with an error. According to the RFC document, this design allows complex distributed workflows to be broken into small, testable, and composable steps.

The Seven-Step Region Migration Workflow

Region migration leverages the procedure framework through a carefully orchestrated sequence:

1. Task Submission: The RegionMigrationManager::submit_region_migration_task function receives a RegionMigrationProcedureTask containing the region ID, source peer, destination peer, timeout, and trigger reason. This creates a unique ProcedureId and registers the procedure with the global ProcedureManager.

2. Concurrency Guarding: Before execution begins, the manager attempts to insert the task into the RegionMigrationProcedureTracker. The insert_running_procedure method returns a RegionMigrationProcedureGuard only if no other migration is running for that region; otherwise, it rejects the duplicate request.

3. State Persistence: The manager builds a RegionMigrationProcedure wrapper that maintains a PersistentContext. After every step, the framework serializes the entire procedure state into the ProcedureStore with Status::Executing { persist: true }, creating a WAL-like layout at /procedures/{PROCEDURE_ID}/{STEP}.step.

4. State Machine Execution: The ProcedureManager drives the migration through states defined in files like migration_start.rs and migration_end.rs. Each state's execute method performs the actual distributed work—such as retrieving region routes or validating leader peers—then returns the next state.

5. Sub-Procedure Handling: When parallel operations are needed (e.g., opening candidate regions on multiple nodes), the current state returns Status::Suspended with a list of sub-procedures. The manager schedules these via ProcedureManager::register_loader and only resumes the parent procedure when all sub-procedures succeed.

6. Crash Recovery: On meta-node restart, the ProcedureManager loads unfinished procedures from ProcedureStore, deserializes them using the registered loader in try_start, and continues from the last persisted step. No migration progress is lost.

7. Result Aggregation: Upon completion, the framework returns a SubmitRegionMigrationTaskResult that summarizes migrated regions, leader changes, and any conflicts encountered.

Concurrency Control and Isolation Mechanisms

The framework prevents race conditions between concurrent administrative commands through a tracker-guard pattern implemented in src/meta-srv/src/procedure/region_migration/manager.rs. The RegionMigrationProcedureTracker maintains a concurrent map of region_id → task entries.

When insert_running_procedure is called, it attempts to insert the task atomically:

impl RegionMigrationProcedureTracker {
    pub(crate) fn insert_running_procedure(
        &self,
        task: &RegionMigrationProcedureTask,
    ) -> Option<RegionMigrationProcedureGuard> {
        let mut map = self.running_procedures.write().unwrap();
        match map.entry(task.region_id) {
            Entry::Occupied(_) => None, // already running → reject
            Entry::Vacant(v) => {
                v.insert(task.clone());
                Some(RegionMigrationProcedureGuard {
                    region_id: task.region_id,
                    running_procedures: self.running_procedures.clone(),
                })
            }
        }
    }
}

The RegionMigrationProcedureGuard implements Drop to automatically remove the entry when the procedure finishes or aborts, guaranteeing that only one migration runs per region at any time. This ensures strict isolation between concurrent operations.

Fault Tolerance and Recovery Mechanisms

Durability is achieved through the PersistentContext structure defined in the procedure framework. After each state transition, the framework serializes the procedure's complete state—including the current step number and all contextual data—to the ProcedureStore. This persistence happens synchronously before the next step executes.

If a meta-node crashes and restarts, the recovery process in ProcedureManager::try_start loads all incomplete procedures from storage. The framework deserializes each procedure using its registered loader and invokes State::next from the last persisted step. Because each state's execute logic is idempotent—producing the same outcome when re-run—the migration safely continues without data corruption or duplicate operations.

Implementation Examples

Submitting a Region Migration Task

The entry point for administrators is the submit_region_migration_task method:

use std::time::Duration;
use common_meta::peer::Peer;
use store_api::storage::RegionId;
use greptime_meta::procedure::region_migration::{
    RegionMigrationTriggerReason, RegionMigrationProcedureTask,
};

let manager = cluster.metasrv.region_migration_manager();
let task = RegionMigrationProcedureTask::new(
    RegionId::new(table_id, region_number),
    Peer::new(from_node_id, from_addr),
    Peer::new(to_node_id, to_addr),
    Duration::from_secs(300),
    RegionMigrationTriggerReason::Manual,
);

let result = manager
    .submit_region_migration_task(task)
    .await
    .expect("failed to submit migration");

println!("Migrated regions: {:?}", result.migrated);

This async call registers the procedure and returns a SubmitRegionMigrationTaskResult while the actual migration executes in the background.

Implementing State Transitions

The RegionMigrationStart state in src/meta-srv/src/procedure/region_migration/migration_start.rs demonstrates how states decide the next step:

async fn next(
    &mut self,
    ctx: &mut Context,
    _procedure_ctx: &ProcedureContext,
) -> Result<(Box<dyn State>, Status)> {
    let mut region_routes = self.retrieve_region_routes(ctx).await?;
    self.filter_unmigrated_regions(&mut region_routes, &ctx.persistent_ctx.to_peer);

    if region_routes.is_empty() {
        return Ok((Box::new(RegionMigrationEnd), Status::done()));
    }

    for route in &region_routes {
        if self.invalid_leader_peer(route, &ctx.persistent_ctx.from_peer)? {
            return Ok((
                Box::new(RegionMigrationAbort::new("Invalid leader")),
                Status::done(),
            ));
        }
    }

    Ok((Box::new(OpenCandidateRegion), Status::executing(true)))
}

This method retrieves metadata, filters already-migrated regions, validates leader assignments, and either proceeds to OpenCandidateRegion or aborts if preconditions fail.

Summary

• The procedure framework provides a generic runtime for fault-tolerant, multi-step distributed operations in GreptimeDB.
• Region migration uses a deterministic state machine (Start → Open Candidate → Upgrade → End) implemented in src/meta-srv/src/procedure/region_migration/.
• The tracker-guard pattern ensures exclusive execution per region, preventing concurrent migration conflicts.
• Persistent snapshots after each step guarantee that migrations survive meta-node crashes and restart from the exact point of interruption.
• Idempotent state logic ensures that re-executing steps after recovery produces consistent, safe results.

Frequently Asked Questions

What happens if a meta-node crashes during region migration?

The migration survives the crash because the ProcedureStore persists the complete procedure state after every step. When the meta-node restarts, the ProcedureManager loads unfinished procedures via try_start, deserializes them using the registered loader, and resumes execution from the last persisted step. No manual intervention is required, and no migration progress is lost.

How does the framework prevent simultaneous migrations of the same region?

The RegionMigrationProcedureTracker in manager.rs maintains an exclusive map of running migrations. The insert_running_procedure method only succeeds if no entry exists for the target region, returning a RegionMigrationProcedureGuard that cleans up on drop. This guarantees that only one migration procedure per region executes at any time, rejecting duplicate submission attempts immediately.

What is the relationship between procedures and sub-procedures?

Sub-procedures allow parallel execution of independent steps. When a state returns Status::Suspended with sub-procedures, the ProcedureManager schedules them concurrently and only resumes the parent procedure when all complete successfully. This enables operations like opening candidate regions on multiple data nodes simultaneously while maintaining deterministic overall orchestration.

How does GreptimeDB ensure migration steps are idempotent?

Each state implementation writes its execute logic to handle re-execution safely. For example, OpenCandidateRegion simply attempts to open the region again if it already exists, yielding the same successful outcome. Combined with persistent state tracking, this idempotency ensures that retries after crashes or network timeouts never corrupt data or leave regions in inconsistent states.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}