# CRRF (Cross Region Replication based Failover) — Design Specification > **Feishu 用户手册**: https://pingcap.feishu.cn/wiki/UdcawlKWYidC4Nk0BrPcfDwbnKh --- ## 1. Overview CRRF enables disaster recovery for TiDB Cloud Dedicated clusters (AWS) by leveraging S3 Cross-Region Replication. When the primary region fails, a new cluster is provisioned in the secondary region using replicated backup data. | Property | Value | |----------|-------| | RPO | ≈ 1 min (kernel checkpoint advancer) | | RTO | ≈ 5 hours (cluster create + snapshot restore + log replay) | | Scope | AWS only (S3 CRR dependency) | | Extra cost | ≈ 10% of original cluster cost | **Design Principles:** - **No new restore flow** — extends existing PiTR restore with region selection + optional compact - **CRRFSetting as SSoT** — no flags on BucketSetting, no duplication - **No feature gate** — controlled via internal API / console access - **Disable is non-destructive** — PiTR and dual-region remain enabled after CRRF disable --- ## 2. Architecture & Topology ### 2.1 Cluster–Bucket Relationship ```mermaid graph TB subgraph "Primary Region (e.g. us-east-1)" UC["Upstream Cluster
(TiDB Dedicated)"] CA["Checkpoint Advancer
(long-running, computes checkpoint)"] MB["Master Bucket
(S3)"] UC -->|"snapshot backup
+ log backup"| MB CA -->|"writes checkpoint
metadata"| MB end subgraph "Secondary Region (e.g. us-west-2)" SB["Secondary Bucket
(S3, CRR replica)"] DC["Downstream Cluster
(created on recovery)"] DCB["Downstream Cluster's
Own Backup Bucket
(created if PiTR enabled)"] SB -->|"snapshot restore +
PiTR log replay"| DC DC -.->|"future backups
(independent lifecycle)"| DCB end MB ==>|"AWS S3 CRR
(async replication)"| SB style UC fill:#4CAF50,color:#fff style CA fill:#8BC34A,color:#fff style DC fill:#FF9800,color:#fff style MB fill:#2196F3,color:#fff style SB fill:#2196F3,color:#fff style DCB fill:#9E9E9E,color:#fff ``` **Key relationships:** - **Master Bucket** (primary region): receives snapshot backups + PiTR log backups from the upstream cluster, and checkpoint metadata from checkpoint advancer - **Secondary Bucket** (secondary region): exact replica via S3 CRR, contains all data needed for recovery - **Downstream Cluster's Own Bucket**: separate bucket created only when the downstream cluster independently enables PiTR; CRRF compact residual files are cleaned here via truncate ### 2.2 Recovery Data Flow ```mermaid flowchart LR subgraph SecBucket["Secondary Bucket (DR Region)"] SS["Snapshot Backup
(replicated)"] LB["Log Backup Files
(replicated)"] CP["Checkpoint Metadata
(replicated)"] CF["Compacted Files
(new, from compact step)"] end subgraph NewCluster["Downstream Cluster"] R1["1. Snapshot Restore"] R2["2. Log Compaction
(optional)"] R3["3. PiTR Log Replay
(to target timestamp)"] end SS --> R1 LB --> R2 R2 --> CF CF --> R3 LB --> R3 R1 --> R2 R2 --> R3 ``` ### 2.3 CRRF Lifecycle State Machine ```mermaid stateDiagram-v2 [*] --> disabled disabled --> enabling : EnableCRRF enabling --> enabled : enable DAG success enabling --> failed : enable DAG start failure failed --> enabling : EnableCRRF (retry) enabled --> disabling : DisableCRRF disabling --> disabled : disable DAG success disabling --> failed : disable DAG start failure ``` **Status definitions:** | Status | Meaning | |--------|---------| | `disabled` | CRRF not configured | | `enabling` | `enable_crrf_v1.0` DAG in progress | | `enabled` | CRRF fully active; cluster is protected | | `disabling` | `disable_crrf_v1.0` DAG in progress | | `failed` | enable or disable DAG failed to start (see `ErrMsg` for details) | `IsActive()` returns `true` for `enabling`, `enabled`, and `disabling` — this is used by guard logic to block conflicting operations. ### 2.4 Entity Relationship Diagram ```mermaid erDiagram CRRFSetting { bigint id PK bigint cluster_id UK bigint project_id bigint tenant_id varchar status varchar secondary_region bigint master_bucket_id FK bigint secondary_bucket_id FK text err_msg } BackupSetting { bigint id PK bigint cluster_id UK bigint project_id bigint tenant_id varchar status varchar infra_provider_type json bucket "BucketSetting (dual-region config, bucket list)" } LogBackup { bigint id PK bigint cluster_id bigint bucket_id FK varchar checkpoint_ts varchar status } Bucket { bigint id PK varchar cluster_id varchar name varchar regions varchar provider varchar status } Restore { bigint id PK bigint cluster_id bigint backup_id FK bigint log_backup_id FK bigint bucket_id FK varchar region "bucket selector (Phase 2)" varchar status } Backup { bigint id PK bigint cluster_id bigint bucket_id FK varchar status } CRRFSetting ||--|| BackupSetting : "cluster_id (logical)" CRRFSetting }o--o| Bucket : "master_bucket_id" CRRFSetting }o--o| Bucket : "secondary_bucket_id" BackupSetting ||--o{ LogBackup : "cluster_id" LogBackup }|--|| Bucket : "bucket_id" Restore }o--|| Backup : "backup_id" Restore }o--o| LogBackup : "log_backup_id" Restore }|--|| Bucket : "bucket_id" Backup }|--|| Bucket : "bucket_id" ``` **Key relationships:** - `CRRFSetting` ↔ `BackupSetting`: logical 1:1 via `cluster_id` (no physical FK; guards query both) - `CRRFSetting.MasterBucketID` / `SecondaryBucketID`: cached bucket IDs populated by `crrf_enable_dual_region` `RunAfter`; zero until dual-region is ready - `LogBackup.CheckpointTS`: used by `QueryCRRF` to derive `RecoverableTime` and `LagSeconds` - `Restore.Region`: new Phase 2 field (already on entity) — selects which region's bucket to restore from --- ## 3. Data Model ### 3.1 New Entity: `CRRFSetting` ```go // app/repo/entity/crrf.go type CRRFSettingStatus string const ( CRRFSettingStatusDisabled CRRFSettingStatus = "disabled" CRRFSettingStatusEnabling CRRFSettingStatus = "enabling" CRRFSettingStatusEnabled CRRFSettingStatus = "enabled" CRRFSettingStatusDisabling CRRFSettingStatus = "disabling" CRRFSettingStatusFailed CRRFSettingStatus = "failed" ) type CRRFSetting struct { Updatable Versioned ClusterID ID `gorm:"type:bigint(20) unsigned;not null;uniqueIndex"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;index"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;index"` Status CRRFSettingStatus `gorm:"type:varchar(32);not null;default:'disabled'"` SecondaryRegion string `gorm:"type:varchar(64);not null;default:''"` // Cached bucket IDs from BackupSetting.Bucket.Buckets after dual-region is ready. // Simplifies downstream queries — avoids re-parsing BucketSetting JSON each time. MasterBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` SecondaryBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` // Error tracking for enable/disable workflow start failures. ErrMsg *string `gorm:"type:text"` } func (CRRFSetting) TableName() string { return "crrf_settings" } // IsActive returns true when CRRF is in a transitional or active state. // Guards use this to block conflicting operations. func (s *CRRFSetting) IsActive() bool { return s.Status == CRRFSettingStatusEnabling || s.Status == CRRFSettingStatusEnabled || s.Status == CRRFSettingStatusDisabling } ``` `CRRFSetting` also has an extension method in `app/repo/entity/crrf_ext.go`: ```go // app/repo/entity/crrf_ext.go // Disable transitions from disabling → disabled and clears all fields. // Returns (false, nil) if already disabled. Returns an error if status is not disabling. func (s *CRRFSetting) Disable(ctx context.Context) (changed bool, err error) { if s.Status == CRRFSettingStatusDisabled { return false, nil } if s.Status != CRRFSettingStatusDisabling { return false, fmt.Errorf("crrf setting status is not disabling, cannot disable, current: %s", s.Status) } s.Status = CRRFSettingStatusDisabled s.SecondaryRegion = "" s.MasterBucketID = 0 s.SecondaryBucketID = 0 s.ErrMsg = nil return true, nil } ``` **Status transitions:** | Current | Action | Next | |---------|--------|------| | `disabled` / `failed` | `EnableCRRF()` | `enabling` | | `enabling` | enable DAG completes successfully | `enabled` | | `enabling` | enable DAG **fails to start** | `failed` (ErrMsg set) | | `enabled` | `DisableCRRF()` | `disabling` | | `disabling` | disable DAG completes successfully | `disabled` | | `disabling` | disable DAG **fails to start** | `failed` (ErrMsg set) | **On disable DAG final step** (`mark_disabled`): `SecondaryRegion` is cleared, `MasterBucketID`/`SecondaryBucketID` zeroed, `ErrMsg` cleared. ### 3.2 Domain Types ```go // app/models/domain/crrf.go type EnableCRRFReq struct { ClusterID uint64 TenantID uint64 ProjectID uint64 SecondaryRegion string // target DR region, e.g. "us-west-2" } type CRRFInfo struct { ClusterID uint64 Status string // "disabled" | "enabling" | "enabled" | "disabling" | "failed" SecondaryRegion string MasterBucketID uint64 SecondaryBucketID uint64 // Derived from LogBackup.CheckpointTS (only when status == "enabled"). CheckpointTS string // latest recoverable point (TSO string) RecoverableTime string // human-readable wall-clock time (RFC3339 UTC) LagSeconds int64 // seconds behind real-time // Non-empty when the last enable/disable attempt failed to start. ErrorMessage string } type DisableCRRFReq struct { ClusterID uint64 TenantID uint64 ProjectID uint64 } ``` ### 3.3 Restore Domain Modifications *(Phase 2 — not yet implemented)* Add a **new `Region` field** and an optional `PiTRSettings` struct to `CreateRestoreReqV2`. **Important**: `Region` is a NEW field, distinct from the existing `BackupRegion`. `BackupRegion` is metadata describing the backup's origin region. The new `Region` field is a **bucket selector** — it tells the restore logic which region's bucket to pull data from. ```go // app/models/domain/restorev2.go (modified) type CreateRestoreReqV2 struct { // ... all existing fields unchanged ... // BackupRegion string ← existing, unchanged — backup's origin region (metadata) // NEW: Region specifies which region's bucket to use for restore. // When dual-region is enabled, backups exist in both the master and // secondary region buckets. This field selects which one to use. // During restore, the system reads BackupSetting.Bucket, iterates // Bucket.Buckets[].Regions, and matches against this value to resolve // the target bucket ID and storage path. // When empty, existing behavior is preserved (uses master bucket). Region string // NEW: Optional PiTR-specific restore settings. PiTRSettings *PiTRSettings } // PiTRSettings contains optional configuration for PiTR restore. type PiTRSettings struct { // CompactBeforePiTR: if true, run log compaction on the log backup files // before PiTR log replay. This converts incremental logs into SST files // for faster restore. Compact runs against the same bucket as the restore. CompactBeforePiTR bool } ``` **Bucket resolution logic** (in restore model): ```go // pseudocode for resolving bucket by Region func resolveBucketByRegion(bucketSetting *entity.BucketSetting, region string) *entity.BackupSettingBucket { if region == "" { return nil // use default (master bucket) } for _, b := range bucketSetting.Buckets { for _, r := range b.Regions { if r.Name == region { return &b } } } return nil // not found — caller should return error } ``` ### 3.4 Repo Interface ```go // app/repo/crrf/crrf_setting_repo.go type CRRFSettingRepo interface { Create(ctx context.Context, setting *entity.CRRFSetting) bizerr.BizError GetByClusterID(ctx context.Context, clusterID uint64) (*entity.CRRFSetting, bizerr.BizError) // Save persists changes with optimistic locking (internally calls entity.SaveWithVersioning). Save(ctx context.Context, setting *entity.CRRFSetting) bizerr.BizError } ``` Implementation uses GORM + `entity.SaveWithVersioning` internally, consistent with other repos in `app/repo/impl/`. ### 3.5 CRRF Configuration (`BackupConfig.CRRFOverrides`) CRRF daemon image overrides are controlled via `BackupConfig.CRRFOverrides` in the service config (`internal/config/config.go`). This allows per-cluster, per-tenant, or per-region image overrides without a code deploy — important for canary rollouts of new daemon versions. ```go // internal/config/config.go (under BackupConfig) CRRFOverrides CRRFOverrides `mapstructure:"crrfOverrides"` // CRRFOverride applies the given image to any cluster matching the FeatureGate. type CRRFOverride struct { Gate FeatureGate `mapstructure:"gate"` Image string `mapstructure:"image"` } type CRRFOverrides []CRRFOverride ``` **`FeatureGate` matching** (priority: Required > Disabled > Enabled > All): | Field | Meaning | |-------|---------| | `requiredProviders` / `requiredRegions` | Match immediately if any entry equals cluster's provider / region | | `disabledProviders/Regions/Tenants/Projects/Clusters` | Exclude if matched | | `enabledProviders/Regions/Tenants/Projects/Clusters` | Include if matched | | `all: true` | Match everything (lowest priority) | **How it is applied** (`overrideCRRFEnableRequest` in `app/workflow/action/crrf/enable/enable_crrf_daemon.go`): ```go for _, override := range config.Global().BackupConfig.CRRFOverrides { if _, enabled := override.Gate.Enabled(config.FeatureGateQueryParams{ Provider: actCtx.MasterRegion.CloudProvider, Region: actCtx.MasterRegion.Name, TenantID: setting.TenantID.Uint64(), ProjectID: setting.ProjectID.Uint64(), ClusterID: setting.ClusterID.Uint64(), }); enabled { if override.Image != "" { req.Image = override.Image // override daemon image } break // first match wins } } ``` **Behavior:** - Iterates `CRRFOverrides` in order; **first matching gate wins** - If `Image` is non-empty, it overrides `EnableCRRFRequest.Image` sent to cluster-server - If no override matches, the default image (from cluster-server's own config) is used - Override failures are **non-fatal** (logged as warnings); the enable flow continues **Example config YAML:** ```yaml backupConfig: crrfOverrides: # Canary: use dev image for a specific test cluster - gate: enabledClusters: [123456] image: "crrf-daemon:dev-20260401" # Rollout: apply new image to us-east-1 before global rollout - gate: requiredRegions: ["us-east-1"] image: "crrf-daemon:v1.2.0" # Default override for all AWS clusters - gate: enabledProviders: ["aws"] image: "crrf-daemon:v1.1.0" ``` --- ## 4. CRRF Model CRRF logic lives in a **dedicated `CRRFModel`** — it is **not** added to `BackupSettingModel`. This keeps concerns separated: `BackupSettingModel` handles BR settings; `CRRFModel` handles CRRF lifecycle. ```go // app/models/crrf/interface.go type CRRFModel interface { EnableCRRF(ctx context.Context, req *domain.EnableCRRFReq) bizerr.BizError QueryCRRF(ctx context.Context, clusterID uint64) (*domain.CRRFInfo, bizerr.BizError) DisableCRRF(ctx context.Context, req *domain.DisableCRRFReq) bizerr.BizError } ``` ### 4.1 EnableCRRF **Preconditions:** 1. `SecondaryRegion` is non-empty 2. BackupSetting exists, status = `normal` 3. Cloud provider = AWS or EKS (rejects GCP/Azure) 4. Existing CRRFSetting `IsActive()` is `false` (i.e., status is `disabled` or `failed`) 5. If dual-region is already enabled: - Secondary region **must match** `req.SecondaryRegion` - If mismatch → **error**: "dual-region is enabled with region X; disable dual-region first or use the same region" 6. If dual-region is not enabled: - Validation happens in the DAG's `enable_dual_region` step **Flow:** 1. Load BackupSetting, validate preconditions 2. Load or create CRRFSetting (reuse existing `disabled`/`failed` record if present) 3. Set `CRRFSetting.Status = enabling`, `SecondaryRegion = req.SecondaryRegion`, clear `ErrMsg` 4. If dual-region already enabled with matching region: - Populate `MasterBucketID` / `SecondaryBucketID` from `BackupSetting.Bucket.Buckets` immediately 5. Save or create CRRFSetting 6. Start `enable_crrf_v1.0` DAG with vars: `crrf_setting_id`, `backup_setting_id`, `cluster_id`, `secondary_region` 7. On DAG start failure: call `markCRRFFailedWithError()` — sets `status = failed`, `ErrMsg` records failure reason **Error handling:** - Enable DAG start failure → `CRRFSetting.Status = failed`, `ErrMsg` records failure reason (caller can retry by issuing `EnableCRRF` again) - Mid-DAG failures are handled by the DAG engine's retry/timeout mechanism ### 4.2 QueryCRRF **Flow:** 1. Load CRRFSetting by clusterID 2. If not found → return `CRRFInfo{Status: "disabled"}` 3. Always return `Status`, `SecondaryRegion`, `MasterBucketID`, `SecondaryBucketID`, `ErrorMessage` 4. If `enabled`: additionally derive recovery point info: - Iterate `LogBackup` list for this cluster; find first entry with non-empty `CheckpointTS` - Parse TSO → physical milliseconds: `physicalMs = tso >> 18` - Set `RecoverableTime = checkpointTime.UTC()` (RFC3339) - Set `LagSeconds = (now - checkpointTime).Milliseconds() / 1000` 5. Return `CRRFInfo` **Note**: There is no separate monitor worker. Recovery point info is derived on-demand from the existing `LogBackup.CheckpointTS`. > **TODO**: Checkpoint is currently sourced from `LogBackup.CheckpointTS` (maintained by the checkpoint advancer and log backup sync worker). A future refactor will source this from a dedicated binary instead. ### 4.3 DisableCRRF **Preconditions:** 1. CRRFSetting exists and is not already `disabled` **Flow:** 1. Load CRRFSetting, validate it exists and is not `disabled` 2. Transition: `Status = disabling` 3. Save CRRFSetting with versioning 4. Start `disable_crrf_v1.0` DAG with vars: `crrf_setting_id`, `cluster_id` 5. On DAG start failure: call `markCRRFFailedWithError()` (sets `status = failed`) **DAG steps** (4 steps): 1. `crrf_toggle_periodic_cluster_settings_backup_disable` (5 min) — disables CRRF periodic backup via management service 2. `crrf_disable_daemon` (15 min) — calls cluster-server to destroy the CRRF daemon; polls until DISABLED/404 3. `crrf_disable_rtc` (5 min) — disables S3 Replication Time Control on the replication rule 4. `crrf_mark_disabled` (5 min) — calls `setting.Disable()`: transitions `disabling → disabled`, clears `SecondaryRegion`, zeroes bucket IDs, clears `ErrMsg` **Important**: PiTR and dual-region remain enabled after CRRF disable — they are independently managed by the user. **DisableCRRF is asynchronous** — it starts a DAG and returns immediately. Poll `QueryCRRF` to track progress. --- ## 5. Restore Modifications (Region Selection + Compact) *(Phase 2 & 3 — not yet implemented)* These are **general enhancements** to the PiTR restore flow, not CRRF-specific. CRRF recovery is simply a PiTR restore call with `Region = secondaryRegion` and `PiTRSettings.CompactBeforePiTR = true`. ### 5.1 Region-Based Bucket Resolution **Problem**: Currently, `running_restore.go` constructs the storage path from `logBackup.Storage` / `logBackup.BucketSettingSnapshot.Prefix`, which always points to the master (primary) bucket. When dual-region is enabled, backups exist in both regions' buckets. **Solution**: Add a new `Region` field to `CreateRestoreReqV2`. When `Region` is set: 1. In the restore model, load `BackupSetting.Bucket` for the cluster 2. Iterate `Bucket.Buckets[].Regions` to find the bucket whose region matches `Region` 3. Use that bucket's ID and storage path instead of the default master bucket **Note**: `Region` is separate from the existing `BackupRegion` field. `BackupRegion` is backup origin metadata; `Region` is a bucket selector for restore. **Files to modify:** | File | Change | |------|--------| | `app/models/domain/restorev2.go` | Add `Region` field and `PiTRSettings` struct to `CreateRestoreReqV2` | | `app/models/restore/impl/pitr_restore.go` | In bucket resolution logic, if `Region` is set, select matching bucket from `BucketSetting.Buckets[]` instead of always using master | | `app/models/restore/impl/pitr_restore.go` — `makeCreateRestoreEntity` | Set `BucketID` to the matched bucket's ID when `Region` is specified | | `app/workflow/action/restore/pitr/create/running_restore.go` | When building `LogBackupPrefix` / `LogBackupStorage`, use the resolved bucket's info instead of the default `logBackup.Storage` path | **Example change in `running_restore.go`** (~L95-105): ```go // Current: always uses logBackup.BucketSettingSnapshot.Prefix (master bucket) restoreParam.BucketInfo.LogBackupPrefix = fmt.Sprintf("%s/%s", *logBackup.BucketSettingSnapshot.Prefix, logBackup.CRName()) // Enhanced: resolve bucket based on Region resolvedBucket := resolveBucketByRegion(logBackup.BucketSettingSnapshot, restore.Region) if resolvedBucket != nil { // Load the matched bucket to get its prefix bucket, _ := bucketRepo.Load(ctx, resolvedBucket.BucketId) restoreParam.BucketInfo.LogBackupPrefix = fmt.Sprintf("%s/%s", bucket.Prefix, logBackup.CRName()) } else { // Fallback to existing behavior (master bucket) restoreParam.BucketInfo.LogBackupPrefix = fmt.Sprintf("%s/%s", *logBackup.BucketSettingSnapshot.Prefix, logBackup.CRName()) } ``` ### 5.2 Compact Before PiTR (Conditional DAG Step) When `PiTRSettings.CompactBeforePiTR == true`, a log compaction step is inserted into the PiTR restore workflow **between** snapshot restore and log replay. **How it works:** 1. The compact step runs against the same bucket as the restore (i.e., the secondary bucket if `Region` points to it) 2. Reuses existing `CompactBackupModel` / compact workflow actions 3. Compact converts incremental log files into SST files → faster PiTR replay 4. Compact produces new files in the bucket — these are cleaned up when the downstream cluster enables PiTR and the next truncate cycle runs **DAG modification to `pit_restore` workflow:** ``` Current PiTR restore DAG: check_and_create_cluster → init_restore_config → add_bucket_policy → run_pit_restoring → restore_sql_bindings → cleanup → final_step Enhanced (when CompactBeforePiTR=true): check_and_create_cluster → init_restore_config → add_bucket_policy → run_snapshot_restore → run_compact_backup → wait_compact_done → run_pitr_log_replay → restore_sql_bindings → cleanup → final_step ``` **Implementation options:** - **Option A**: Modify the existing PiTR restore DAG to conditionally include the compact step (check a DAG variable) - **Option B**: Create a separate DAG variant `pit_restore_with_compact_v1.0` — model selects the appropriate DAG based on `PiTRSettings` Option B is simpler and avoids complicating the existing well-tested DAG. Recommend Option B. **Passing Region through the DAG**: The `Region` value needs to be stored on the `Restore` entity (or as a DAG variable) so workflow actions can access it. Add a field to `entity.Restore`: ```go // entity.Restore — add field Region *string `gorm:"type:varchar(64)"` // which region's bucket to use for restore ``` --- ## 6. Guard Logic (CRRF Lock) When CRRF `IsActive()` (i.e., `enabling`, `enabled`, or `disabling`), users must not independently disable PiTR or dual-region. ### 6.1 PiTR Disable Guard **Location**: `app/models/backup/impl/backup_setting.go` — `preCheckModifyPitBackupSetting()` ```go // When req.Enable == false (disabling PiTR): crrfSetting, _ := m.crrfSettingRepo.GetByClusterID(ctx, req.ClusterID) if crrfSetting != nil && crrfSetting.IsActive() { return bizerr.New(errcode.ErrCodeParams, "cannot disable PiTR while CRRF is active; disable CRRF first") } ``` ### 6.2 Dual-Region Disable Guard **Location**: `app/models/backup/impl/region_modification.go` — both `awsDisableDualRegionHook` and `gcpRegionModifierHook` `paramCheck()` ```go // When disabling dual-region: crrfSetting, _ := m.crrfSettingRepo.GetByClusterID(ctx, req.ClusterID) if crrfSetting != nil && crrfSetting.IsActive() { return bizerr.New(errcode.ErrCodeParams, "cannot disable dual-region while CRRF is active; disable CRRF first") } ``` ### 6.3 SSoT Principle Guards query `CRRFSetting` directly via `IsActive()` — there is **no** `CRRFEnable` flag on `BucketSetting`. This avoids data duplication and inconsistency. --- ## 7. Workflows ### 7.1 enable_crrf DAG (`enable_crrf_v1.0`) **Location**: `app/workflow/dag/crrf/enable.go` **DAG variables**: `crrf_setting_id`, `backup_setting_id`, `cluster_id`, `secondary_region` **Actions location**: `app/workflow/action/crrf/enable/`, `app/workflow/action/crrf/common/` Each action uses fastflow's `Run` / `RunAfter` pattern: `Run` triggers the operation (idempotent), `RunAfter` polls for completion. ```mermaid flowchart TD A["1. crrf_enable_dual_region
(40min timeout)
Run: trigger ModifyRegionSetting(enable=true)
RunAfter: poll until active, cache bucket IDs"] --> B["2. crrf_enable_rtc
(5min timeout)
Run: enable S3 RTC on replication rule"] B --> C["3. crrf_enable_pitr
(40min timeout)
Run: trigger ModifyPitBackupSetting(enable=true)
RunAfter: poll until LogBackup status=running"] C --> D["4. crrf_enable_daemon
(15min timeout)
Run: crrfFacade.EnableCRRF(clusterID, crrfId, secondaryBucketURI)
RunAfter: poll GetCRRF until RUNNING"] D --> E["5. crrf_toggle_periodic_cluster_settings_backup_enable
(5min timeout)
Run: EnablePeriodicBackup(clusterID)"] E --> F["6. crrf_mark_enabled
(5min timeout)
Run: CRRFSetting.Status = 'enabled'"] ``` **Step details:** | Step | Action name | Timeout | What it does | |------|------------|---------|--------------| | 1 | `crrf_enable_dual_region` | 40 min | `Run`: calls `BackupSettingModel.ModifyRegionSetting(enable=true, [master, secondary])` — idempotent if already enabled. `RunAfter`: polls `BackupSetting` until `DualRegionEnable=true` and `Status=normal`, then saves bucket IDs into `CRRFSetting`. | | 2 | `crrf_enable_rtc` | 5 min | `Run`: loads `CRRFActionContext`. Calls `awsBucketManager.ModifyReplicationRTC` on the master bucket's replication rule with `EnableRTC: true`. Rule ID is derived from `BackupSetting.Bucket.Prefix`. | | 3 | `crrf_enable_pitr` | 40 min | `Run`: calls `BackupSettingModel.ModifyPitBackupSetting(enable=true)` — idempotent if already enabled. `RunAfter`: polls `LogBackup` until a `running` log backup is found. | | 4 | `crrf_enable_daemon` | 15 min | `Run`: loads `CRRFActionContext` (master region, CRRFSetting, LogBackup name). Resolves secondary bucket URI from `SecondaryBucketID`. Calls `crrfFacade.EnableCRRF(clusterId, logBackupName, crrfId="crrf-{settingID}", downstreamStorage=uri)`. Skip if daemon already RUNNING/PENDING. `RunAfter`: polls `GetCRRF` until phase=RUNNING; `StopRetry` on FAILED. | | 5 | `crrf_toggle_periodic_cluster_settings_backup_enable` | 5 min | Calls `CRRFPeriodicBackupClient.EnablePeriodicBackup(clusterID)`. No `RunAfter`. | | 6 | `crrf_mark_enabled` | 5 min | Loads `CRRFSetting` by `cluster_id` and transitions `enabling → enabled`. Idempotent; warns and skips if status is unexpected. | **Failure handling**: On any step timeout or unrecoverable error, the DAG engine marks the instance failed. `CRRFSetting` remains in `enabling` state until the next `EnableCRRF` call retries. The caller can reissue `EnableCRRF` to start a fresh DAG. ### 7.2 disable_crrf DAG (`disable_crrf_v1.0`) **Location**: `app/workflow/dag/crrf/disable.go` **DAG variables**: `crrf_setting_id`, `cluster_id` **Actions location**: `app/workflow/action/crrf/disable/`, `app/workflow/action/crrf/common/` ```mermaid flowchart TD A["1. crrf_toggle_periodic_cluster_settings_backup_disable
(5min timeout)
Run: DisablePeriodicBackup(clusterID)"] --> B["2. crrf_disable_daemon
(15min timeout)
Run: crrfFacade.DisableCRRF(clusterId, crrfId)
RunAfter: poll until DISABLED or 404"] B --> C["3. crrf_disable_rtc
(5min timeout)
Run: disable S3 RTC on replication rule"] C --> D["4. crrf_mark_disabled
(5min timeout)
Run: setting.Disable() → Status='disabled', clear fields"] ``` **Step details:** | Step | Action name | Timeout | What it does | |------|------------|---------|--------------| | 1 | `crrf_toggle_periodic_cluster_settings_backup_disable` | 5 min | Calls `CRRFPeriodicBackupClient.DisablePeriodicBackup(clusterID)`. | | 2 | `crrf_disable_daemon` | 15 min | `Run`: loads `CRRFActionContext`. Checks `GetCRRF` — if 404 or already DISABLED/DESTROYING → skip. Calls `crrfFacade.DisableCRRF(clusterId, logBackupName, crrfId)`. `RunAfter`: polls `GetCRRF` until phase=DISABLED or 404; `StopRetry` on FAILED. | | 3 | `crrf_disable_rtc` | 5 min | `Run`: loads `CRRFActionContext`. Calls `awsBucketManager.ModifyReplicationRTC` on the master bucket's replication rule with `EnableRTC: false`. Rule ID is derived from `BackupSetting.Bucket.Prefix`. | | 4 | `crrf_mark_disabled` | 5 min | Calls `setting.Disable()` (validates `disabling → disabled`, clears `SecondaryRegion`, zeroes bucket IDs, clears `ErrMsg`). Saves. Idempotent (no-op if already disabled). | **Each step is idempotent** — safe to retry on failure. ### 7.3 CRRF Daemon (Cluster-Server Integration) The CRRF daemon is a component managed by the **cluster-server** (regional API) that handles the actual data replication. The dataflow service interacts with it via `pkg/regional-api/`. **Daemon lifecycle phases** (`CRRFPhase` in `pkg/regional-api/crrf_types.go`): | Phase | Meaning | |-------|---------| | `PENDING` | Daemon is being provisioned | | `RUNNING` | Daemon is active and replicating | | `DESTROYING` | Daemon is being shut down | | `FAILED` | Daemon failed | | `DISABLED` | Daemon is stopped | **Cluster-server API endpoints** (called via `crrfFacade` / `pkg/regional-api/crrf.go`): ``` POST /cluster-server/api/v1/clusters/{cluster_id}/logbackups/{log_backup_name}:enableCRRF POST /cluster-server/api/v1/clusters/{cluster_id}/logbackups/{log_backup_name}:disableCRRF GET /cluster-server/api/v1/clusters/{cluster_id}/logbackups/{log_backup_name}:getCRRF ``` **`EnableCRRFRequest`** fields: `clusterId`, `logBackupName`, `crrfId`, `stateStorageSubDir`, `downstreamStorage`, `image` **Daemon ID**: `CRRFDaemonID(settingID) = "crrf-{settingID}"` — stable and deterministic; shared between enable and disable actions. **`CRRFActionContext`** (`app/workflow/action/crrf/common/load_crrf_action_context.go`) — shared context loaded by all CRRF workflow actions: - `MasterRegion *regionalapi.Region` — derived from BackupSetting (TenantID, Region name, CloudProvider) - `CRRFSetting *entity.CRRFSetting` — loaded via `crrfSettingRepo.GetByClusterID` - `BackupSetting *entity.BackupSetting` — loaded via `backupSettingRepo.GetByFilter`; provides `Bucket.Prefix` (replication rule ID) and `MasterRegion()` - `LogBackupName string` — K8s CR name of the first running log backup (from `FindRunningLogBackupCRName`) ### 7.4 CRRFPeriodicBackupClient `CRRFPeriodicBackupClient` is an interface injected into the toggle action. It wraps the management service API for cluster settings periodic backup. ```go // app/models/backup interface (referenced by container) type CRRFPeriodicBackupClient interface { EnablePeriodicBackup(ctx context.Context, clusterID uint64) bizerr.BizError DisablePeriodicBackup(ctx context.Context, clusterID uint64) bizerr.BizError } ``` The container currently wires `&backup.NoopCRRFPeriodicBackupClient{}` (a no-op stub that logs calls but does nothing) until the real management service integration is implemented. > **TODO**: Replace `NoopCRRFPeriodicBackupClient` with real management service API call. --- ## 8. Controller Layer ### 8.1 Endpoints CRRF management uses dedicated endpoints. CRRF recovery uses the **existing restore endpoint** with enhanced parameters (Phase 2). ``` POST /api/v1/orgs/{org_id}/projects/{project_id}/clusters/{cluster_id}/crrf → EnableCRRF GET /api/v1/orgs/{org_id}/projects/{project_id}/clusters/{cluster_id}/crrf → QueryCRRF DELETE /api/v1/orgs/{org_id}/projects/{project_id}/clusters/{cluster_id}/crrf → DisableCRRF POST /clusters/{clusterID}/restores → (existing, Phase 2) with Region + PiTRSettings ``` **Controller location**: `app/controllers/console/crrf.go` The `EnableCRRFResponse` and `DisableCRRFResponse` both return `status` and `message` immediately. Since both operations are async, poll `QueryCRRF` for progress. ### 8.2 Proto Definition **Location**: `idl/idl/dataflow/dataflow/crrf.proto` ```protobuf service CRRFConsoleService { rpc EnableCRRF (EnableCRRFRequest) returns (EnableCRRFResponse) // POST rpc QueryCRRF (QueryCRRFRequest) returns (QueryCRRFResponse) // GET rpc DisableCRRF (DisableCRRFRequest) returns (DisableCRRFResponse) // DELETE } enum CRRFStatus { CRRF_STATUS_UNSPECIFIED = 0; CRRF_STATUS_DISABLED = 1; CRRF_STATUS_ENABLING = 2; CRRF_STATUS_ENABLED = 3; CRRF_STATUS_DISABLING = 4; CRRF_STATUS_FAILED = 5; } message EnableCRRFRequest { uint64 org_id = 1; uint64 project_id = 2; uint64 cluster_id = 3; string secondary_region = 4; } message EnableCRRFResponse { CRRFStatus status = 1; // ENABLING string message = 2; } message QueryCRRFRequest { uint64 org_id = 1; uint64 project_id = 2; uint64 cluster_id = 3; } message QueryCRRFResponse { uint64 cluster_id = 1; CRRFStatus status = 2; string secondary_region = 3; uint64 master_bucket_id = 4; uint64 secondary_bucket_id = 5; string checkpoint_ts = 6; // latest recoverable point (TSO) string recoverable_time = 7; // human-readable wall-clock time (RFC3339 UTC) int64 lag_seconds = 8; // replication lag string error_message = 9; // non-empty when last op failed to start } message DisableCRRFRequest { uint64 org_id = 1; uint64 project_id = 2; uint64 cluster_id = 3; } message DisableCRRFResponse { CRRFStatus status = 1; // DISABLING string message = 2; } ``` The existing `CreateRestoreRequest` proto gains (Phase 2): ```protobuf string region = N; // optional — bucket region selector (NOT BackupRegion) PiTRSettings pitr_settings = M; // optional message PiTRSettings { bool compact_before_pitr = 1; } ``` --- ## 9. DI Container ```go // app/container/container.go — CRRF-related additions type GlobalContainer struct { // ... existing fields ... CRRFSettingRepo crrf_repo.CRRFSettingRepo CRRFModel crrf_model.CRRFModel CRRFPeriodicBackupClient backup.CRRFPeriodicBackupClient // stub (NoopCRRFPeriodicBackupClient) CRRFController console_api.CRRFController } // initRepos: container.CRRFSettingRepo = crrf_repo_impl.NewCRRFSettingRepo(container.TransProvider) // initModels: container.CRRFModel = crrf_model_impl.NewCRRFModel( container.BackupSettingRepo, container.LogBackupRepo, container.CRRFSettingRepo, container.WorkflowModel, ) container.CRRFPeriodicBackupClient = &backup.NoopCRRFPeriodicBackupClient{} // no-op until real integration // initControllers: container.CRRFController = console_api.NewCRRFController( container.TransProvider, container.CRRFModel, ) ``` **Note**: `BackupSettingModel` also depends on `CRRFSettingRepo` (injected separately) for guard logic in `preCheckModifyPitBackupSetting` and dual-region hooks. --- ## 10. Migration ```go // migrate/record_202604021100_add_crrf_setting.go var _ = AddDataflowDBMigrateRecord( "202604021100", func(db *gorm.DB) error { if err := db.AutoMigrate(&entity.CRRFSetting{}); err != nil { return errors.Annotate(err, "auto migrate 'crrf_settings'") } log.Info("migration 202604021100 completed: add crrf_settings table") return nil }, ) ``` If adding `Region` field to `entity.Restore` (Phase 2): ```go var _ = AddDataflowDBMigrateRecord( "YYYYMMDDHHMM", func(db *gorm.DB) error { if err := db.AutoMigrate(&entity.Restore{}); err != nil { return errors.Annotate(err, "auto migrate 'restores' for region") } log.Info("migration completed: add region to restores") return nil }, ) ``` --- ## 11. Technical Notes ### 11.1 S3 Versioning & Lifecycle - Dual-region enable already configures S3 bucket versioning and lifecycle rules on both master and secondary buckets - Versioning causes object deletes to be soft-deletes → lifecycle rules handle cleanup to prevent cost leakage - **No additional bucket configuration needed for CRRF** ### 11.2 Compact File Cleanup - CRRF recovery runs compact on the secondary bucket, producing new SST files - These files are cleaned up when the downstream cluster enables PiTR and the next `truncate` cycle runs - If the downstream cluster never enables PiTR, these files persist (acceptable cost in a DR scenario) ### 11.3 Data Integrity & Checkpoint - BR does **not** perform data integrity checks during restore - Restorability from the secondary bucket relies on the **Checkpoint Advancer** — a long-running program deployed on the upstream cluster that computes checkpoint timestamps and writes them to S3 - The `CheckpointTS` in the `LogBackup` entity represents the latest point safe to restore to - S3 CRR replicates checkpoint metadata along with backup data → the secondary bucket's checkpoint lags by CRR replication delay - `QueryCRRF` exposes this as `RecoverableTime` and `LagSeconds` ### 11.4 CRRF Recovery as Standard Restore A CRRF recovery is indistinguishable from a standard PiTR restore at the workflow level. The only differences are parameter values: | Parameter | Normal PiTR Restore | CRRF Recovery | |-----------|---------------------|---------------| | `Region` | empty (uses master bucket) | secondary region | | `PiTRSettings.CompactBeforePiTR` | `false` (or nil) | `true` | | `ClusterConfig.Region` | same as backup region | secondary region | This means CRRF recovery benefits from all existing restore infrastructure: progress tracking, cancellation, error handling, retry, etc. --- ## 12. Files Created / Modified ### Phase 1 — New Files (✅ Implemented) | File | Description | |------|-------------| | `app/repo/entity/crrf.go` | `CRRFSetting` entity + 5 status constants + `IsActive()` | | `app/repo/entity/crrf_ext.go` | `CRRFSetting.Disable()` method | | `app/repo/crrf/crrf_setting_repo.go` | `CRRFSettingRepo` interface (`Create`, `GetByClusterID`, `Save`) | | `app/repo/crrf/impl/crrf_setting_repo.go` | GORM repo implementation | | `app/models/domain/crrf.go` | `EnableCRRFReq`, `DisableCRRFReq`, `CRRFInfo` domain types | | `app/models/crrf/interface.go` | `CRRFModel` interface | | `app/models/crrf/impl/crrf.go` | `CRRFModel` implementation | | `app/models/backup/crrf_periodic_backup_client.go` | `CRRFPeriodicBackupClient` interface + `NoopCRRFPeriodicBackupClient` | | `app/workflow/dag/crrf/enable.go` | `enable_crrf_v1.0` DAG definition (6 steps) | | `app/workflow/dag/crrf/disable.go` | `disable_crrf_v1.0` DAG definition (4 steps) | | `app/workflow/action/crrf/common/load_crrf_action_context.go` | Shared `CRRFActionContext` loader | | `app/workflow/action/crrf/common/toggle_periodic_cluster_settings_backup.go` | Step 5 of enable DAG / step 1 of disable DAG | | `app/workflow/action/crrf/enable/enable_dual_region.go` | Step 1 of enable DAG | | `app/workflow/action/crrf/enable/enable_rtc.go` | Step 2 of enable DAG — enables S3 RTC on replication rule | | `app/workflow/action/crrf/enable/enable_pitr.go` | Step 3 of enable DAG | | `app/workflow/action/crrf/enable/enable_crrf_daemon.go` | Step 4 of enable DAG (cluster-server enable) | | `app/workflow/action/crrf/enable/mark_enabled.go` | Step 6 of enable DAG | | `app/workflow/action/crrf/disable/disable_crrf_daemon.go` | Step 2 of disable DAG (cluster-server disable) | | `app/workflow/action/crrf/disable/disable_rtc.go` | Step 3 of disable DAG — disables S3 RTC on replication rule | | `app/workflow/action/crrf/disable/mark_disabled.go` | Step 4 of disable DAG | | `pkg/regional-api/crrf_types.go` | `CRRFPhase` enum + request/response types | | `pkg/regional-api/crrf.go` | CRRF daemon client wrapping cluster-server API | | `app/controllers/console/crrf.go` | `CRRFController` (console API handler) | | `idl/idl/dataflow/dataflow/crrf.proto` | gRPC service + message definitions | | `migrate/record_202604021100_add_crrf_setting.go` | DB migration for `crrf_settings` table | ### Phase 1 — Modified Files (✅ Implemented) | File | Change | |------|--------| | `app/models/backup/impl/backup_setting.go` | CRRF guard in `preCheckModifyPitBackupSetting` | | `app/models/backup/impl/region_modification.go` | CRRF guard in dual-region disable for both AWS and GCP hooks | | `app/workflow/constants.go` | CRRF DAG/action/key constants | | `app/container/container.go` | Wire `CRRFSettingRepo`, `CRRFModel`, `CRRFPeriodicBackupClient`, `CRRFController` | ### Phase 2 — Restore Region Selection *(Not yet implemented)* | File | Change | |------|--------| | `app/models/domain/restorev2.go` | Add `Region` field and `PiTRSettings` struct to `CreateRestoreReqV2` | | `app/repo/entity/entity.go` or `restore.go` | Add `Region` field to `Restore` entity | | `app/models/restore/impl/pitr_restore.go` | Bucket resolution by `Region` | | `app/workflow/action/restore/pitr/create/running_restore.go` | Support secondary bucket storage path | | Migration file | Add `region` column to `restores` table | ### Phase 3 — Compact Before PiTR *(Not yet implemented)* | File | Change | |------|--------| | `app/models/domain/restorev2.go` | Add `PiTRSettings` struct to `CreateRestoreReqV2` | | `app/workflow/dag/restore/restore_pitr_with_compact.go` | PiTR restore DAG variant with compact step | | Proto file | Add `PiTRSettings` message to `CreateRestoreRequest` | --- ## 13. Implementation Phases ### Phase 1: Core CRRF (Enable / Query / Disable) — ✅ Implemented - `CRRFSetting` entity + `Disable()` ext method, repo, migration (`202604021100`) - Dedicated `CRRFModel` (`app/models/crrf/`) - `enable_crrf_v1.0` DAG: **5 steps** (dual-region → PiTR → enable daemon → periodic backup → mark enabled) - `disable_crrf_v1.0` DAG: **3 steps** (disable periodic backup → disable daemon → mark disabled) - CRRF daemon integration via cluster-server regional API (`pkg/regional-api/crrf.go`) - Guard logic in `BackupSettingModel` using `CRRFSetting.IsActive()` - Console controller + proto (`app/controllers/console/crrf.go`, `idl/idl/dataflow/dataflow/crrf.proto`) - 5 lifecycle statuses: `disabled`, `enabling`, `enabled`, `disabling`, `failed` - Both enable and disable DAG start failures → `failed` (ErrMsg set); retry-able by calling `EnableCRRF` again - `CRRFPeriodicBackupClient` interface — wired as no-op stub pending real management service integration ### Phase 2: Restore Region Selection — 🔲 Not yet implemented - Add `Region` field to `CreateRestoreReqV2` (new field, separate from existing `BackupRegion`) - Add `Region` to `Restore` entity + migration - Modify bucket resolution in `pitr_restore.go` and `running_restore.go` - Test with dual-region enabled clusters ### Phase 3: Compact Before PiTR — 🔲 Not yet implemented - Add `PiTRSettings` to `CreateRestoreReqV2` + proto - Create `restore_pitr_with_compact` DAG variant - Implement compact step in restore workflow - Test end-to-end CRRF recovery flow ### Phase 4: Monitoring & Observability — 🔲 Not yet implemented - Grafana dashboard for CRRF status / lag - Alerting on excessive lag - Metering / billing integration - Real `CRRFPeriodicBackupClient` integration (replace no-op stub)