# CRPiTR Per-Region Checkpoint Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Add `groupPiTRTimeRangesByRegionV2` so that secondary-region buckets (`bucket.Master == false` in `BucketSettingSnapshot`) use the actual cross-region checkpoint (from `resume-state.json` in the bucket storage) as the recoverable time-range end, instead of the main region's DB checkpoint. Primary-region buckets (`bucket.Master == true`) continue to use the DB checkpoint unchanged. **Architecture:** A new file `pit_backup_checkpoint.go` (same package as `pit_backup.go`) contains `PersistentState`, `GetCheckpointForCRPiTR` (method on `pitBackupModel`), and `BuildStorageOption`. The `pitBackupModel` struct gains `bucketRepo`, `getStorageFunc` (default `blob.NewStorage`), and `getCheckpointForCRPiTR` (injectable function field, default bound to `m.GetCheckpointForCRPiTR` for testability). A new `groupPiTRTimeRangesByRegionV2` function is added; V1 is kept unchanged. `pit_backup_testhelper.go` (non-test file) exports `NewPitBackupModelWithStorageFunc` for external test packages that need to inject a mock storage function. **Tech Stack:** Go, `github.com/tidbcloud/dataflow-service/pkg/blob` (S3 access), `github.com/tidbcloud/dataflow-service/app/repo` (BucketRepo), standard `encoding/json`. --- ## File Map | File | Action | Responsibility | |------|--------|----------------| | `app/models/backup/impl/pit_backup.go` | Modify | Add fields to struct; update `NewPitBackupModel`; add `groupPiTRTimeRangesByRegionV2` | | `app/models/backup/impl/pit_backup_checkpoint.go` | Create | `PersistentState`, `GetCheckpointForCRPiTR`, `BuildStorageOption` | | `app/models/backup/impl/pit_backup_testhelper.go` | Create | `NewPitBackupModelWithStorageFunc` (exported, non-test file) | | `app/models/backup/impl/pit_backup_test.go` | Create | Unit tests in package `impl` for `mergeSingleRegionPiTRTimeRanges` and `groupPiTRTimeRangesByRegionV2` | | `app/container/container.go` | Modify | Pass `container.BucketRepo` as 6th arg to `NewPitBackupModel` | --- ## Task 1: Add fields to `pitBackupModel` and update `NewPitBackupModel` **Files:** - Modify: `app/models/backup/impl/pit_backup.go` - Modify: `app/container/container.go` - [x] **Step 1: Add fields to struct** In `app/models/backup/impl/pit_backup.go`, update the struct: ```go type pitBackupModel struct { // db dbProvider db.TransProvider pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo logBackupRepo backupRepo.LogBackupRepo backupRepo backupRepo.BackupRepo scheduleRepo backupRepo.BackupScheduleRepo restoreRepo restoreRepo.RestoreRepo bucketRepo bucketRepo.BucketRepo // NEW // sub modules snapshotBackupModel backup.SnapshotBackupModel backupSettingModel backup.BackupSettingModel // internal helpers workflowCli workflow.Workflow coreCli grpcli.CoreGrpcClient switchStatusMatrix status_matrix.StatusMatrix featureGate feature_gate.FeatureGate // injectable for CRPiTR checkpoint and testing getStorageFunc blob.GetStorageFunc // NEW; default: blob.NewStorage getCheckpointForCRPiTR func( // NEW; default: bound to m.GetCheckpointForCRPiTR ctx context.Context, logBackup entity.LogBackup, region string, getStorageFunc blob.GetStorageFunc, ) (int64, bool, error) } ``` Import alias for base repo: ```go bucketRepo "github.com/tidbcloud/dataflow-service/app/repo" ``` - [x] **Step 2: Update `NewPitBackupModel` — add `bucketRepo` as 6th param, two-step init** ```go func NewPitBackupModel( dbProvider db.TransProvider, pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo, logBackupRepo backupRepo.LogBackupRepo, backupRepo backupRepo.BackupRepo, restoreRepo restoreRepo.RestoreRepo, bucketRepo bucketRepo.BucketRepo, // NEW — position 6 workflowCli workflow.Workflow, coreCli grpcli.CoreGrpcClient, snapshotBackupModel backup.SnapshotBackupModel, backupSettingModel backup.BackupSettingModel, scheduleRepo backupRepo.BackupScheduleRepo, featureGate feature_gate.FeatureGate, ) backup.PitBackupModel { m := &pitBackupModel{ dbProvider: dbProvider, pitBackupSwitchRepo: pitBackupSwitchRepo, logBackupRepo: logBackupRepo, backupRepo: backupRepo, scheduleRepo: scheduleRepo, restoreRepo: restoreRepo, bucketRepo: bucketRepo, workflowCli: workflowCli, coreCli: coreCli, snapshotBackupModel: snapshotBackupModel, backupSettingModel: backupSettingModel, switchStatusMatrix: NewPitBackupSwitchStatusMatrix(), featureGate: featureGate, getStorageFunc: blob.NewStorage, } m.getCheckpointForCRPiTR = m.GetCheckpointForCRPiTR // bound after creation return m } ``` > **Why two-step?** `m.GetCheckpointForCRPiTR` is a method value bound to `m`. Go doesn't > allow referencing `m.Method` inside the struct literal because `m` doesn't exist yet. - [x] **Step 3: Update DI call in `container.go`** Pass `container.BucketRepo` at position 6 (after `container.RestoreRepo`): ```go container.PitBackupModel = backup_impl.NewPitBackupModel( container.TransProvider, // 1 container.PitBackupSwitchRepo, // 2 container.LogBackupRepo, // 3 container.BackupRepo, // 4 container.RestoreRepo, // 5 container.BucketRepo, // 6 — NEW container.WorkflowModel, // 7 container.CoreGrpcClient, // 8 container.SnapshotBackupModel, // 9 container.BackupSettingModel, // 10 container.BackupScheduleRepo, // 11 container.FeatureGateV2, // 12 ) ``` - [x] **Step 4: Verify compile** ```bash cd repos/dataflow-service && go build ./app/... ``` Expected: no errors. --- ## Task 2: Create `pit_backup_checkpoint.go` **Files:** - Create: `app/models/backup/impl/pit_backup_checkpoint.go` This file contains three things: 1. `PersistentState` — mirrors `resume-state.json` 2. `GetCheckpointForCRPiTR` — method on `pitBackupModel`; loads bucket from snapshot, reads file from S3 3. `BuildStorageOption` — exported helper that builds `blob.StorageOption` - [x] **Step 1: Create `pit_backup_checkpoint.go`** ```go package impl import ( "context" "encoding/json" "fmt" "github.com/tidbcloud/core-svc/idl/pbgen/proto/core" // only if needed "github.com/tidbcloud/dataflow-service/app/models/domain" "github.com/tidbcloud/dataflow-service/app/repo/entity" "github.com/tidbcloud/dataflow-service/internal/model/util" "github.com/tidbcloud/dataflow-service/pkg/blob" "go.uber.org/zap" ) const crPiTRResumeStateFile = "resume-state.json" // PersistentState mirrors the checkpoint file written by the CRPiTR replicator. // Refer to br/pkg/task/stream.go type PersistentState struct { LastCheckpoint uint64 `json:"last_checkpoint"` } // GetCheckpointForCRPiTR reads resume-state.json from the bucket covering the // given region and returns the checkpoint as Unix seconds. // // Returns 0, false, nil if the bucket/file is not found (CRRF not started). // Returns 0, false, err on read/parse error (caller should log and skip). func (b *pitBackupModel) GetCheckpointForCRPiTR( ctx context.Context, logBackup entity.LogBackup, region string, getStorageFunc blob.GetStorageFunc, ) (int64, bool, error) { storageURI := logBackup.Storage bucketSnapshot, found := logBackup.BucketSettingSnapshot.GetBucketByRegion(region) if !found { return 0, false, nil } bucket, loadErr := b.bucketRepo.Load(ctx, bucketSnapshot.BucketId) if loadErr != nil { return 0, false, fmt.Errorf("failed to load bucket: %w", loadErr) } opt := BuildStorageOption(bucket, region) storage, err := getStorageFunc(ctx, storageURI, opt) if err != nil { if blob.IsNotFoundError(err) { return 0, false, nil } zap.L().Warn("GetCheckpointForCRPiTR: failed to open storage", zap.String("storageURI", storageURI), zap.String("region", region), zap.Error(err)) return 0, false, fmt.Errorf("failed to open storage: %w", err) } data, err := storage.ReadFile(ctx, crPiTRResumeStateFile) if err != nil { if blob.IsNotFoundError(err) { return 0, false, nil } zap.L().Warn("GetCheckpointForCRPiTR: failed to read resume-state.json", zap.String("storageURI", storageURI), zap.String("region", region), zap.Error(err)) return 0, false, fmt.Errorf("failed to read resume-state.json: %w", err) } var state PersistentState if err := json.Unmarshal(data, &state); err != nil { zap.L().Warn("GetCheckpointForCRPiTR: failed to parse resume-state.json", zap.String("storageURI", storageURI), zap.String("region", region), zap.Error(err)) return 0, false, fmt.Errorf("failed to parse resume-state.json: %w", err) } return util.TransToTS(state.LastCheckpoint), true, nil } // BuildStorageOption builds a blob.StorageOption for the given bucket and target region. // For AWS/EKS buckets it sets AWSRoleARN; other providers leave it unset. // Note: function name preserves existing codebase typo ("Stoage") for consistency. func BuildStorageOption(bucket *entity.Bucket, region string) *blob.StorageOption { opt := &blob.StorageOption{ ClusterRegion: region, ProjectID: bucket.Uint64ProjectID(), } if bucket.InfraProviderType == entity.InfraProviderTypeAWS || bucket.InfraProviderType == entity.InfraProviderTypeEKS { if bucket.CloudAccountID != nil { opt.AWSRoleARN = domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID) } } return opt } ``` > **Key design points:** > - `GetCheckpointForCRPiTR` is a method (has receiver `b *pitBackupModel`) so it can access > `b.bucketRepo.Load`. It accepts `getStorageFunc` as a parameter so callers can inject a > mock without patching the struct field. > - `BucketSettingSnapshot.GetBucketByRegion(region)` (on `*entity.BucketSetting`) returns a > `BackupSettingBucket` struct with the `BucketId` field. - [x] **Step 2: Verify compile** ```bash cd repos/dataflow-service && go build ./app/models/backup/impl/... ``` Expected: no errors. --- ## Task 3: Add `groupPiTRTimeRangesByRegionV2` to `pit_backup.go` **Files:** - Modify: `app/models/backup/impl/pit_backup.go` V2 is a **new function** added alongside the unchanged V1. It takes `cluster` and `cloudRegions` as pre-fetched parameters (cleaner for unit tests). It iterates all logbackups uniformly and uses `bucket.Master` to distinguish primary from secondary regions. > **V1 is NOT modified.** `mergePiTRTimeRanges` still calls V1. Wiring V2 into the call chain > is a follow-on step. - [x] **Step 1: Add `groupPiTRTimeRangesByRegionV2`** Append to `pit_backup.go` after the existing `groupPiTRTimeRangesByRegion`: ```go func (b *pitBackupModel) groupPiTRTimeRangesByRegionV2( ctx context.Context, logBackupTimeRanges []*domain.PiTRTimeRange, cluster *core.Cluster, cloudRegions map[string]map[string]*core.CloudRegion, ) ([]*domain.PiTRRegionTimeRange, error) { logger := log.GetLogger(ctx) regions := cloudRegions[cluster.CloudProviderName] timeRangeMap := make(map[string]*domain.PiTRRegionTimeRange) for _, logBackupTimeRange := range logBackupTimeRanges { logBackup := logBackupTimeRange.LogBackup for _, bucket := range logBackup.BucketSettingSnapshot.Buckets { var endTs int64 if bucket.Master { endTs = logBackupTimeRange.EndTS } else { checkpoint, found, err := b.getCheckpointForCRPiTR( ctx, logBackup, bucket.Regions[0], b.getStorageFunc) if err != nil { logger.Error("get checkpoint for CR PiTR error", zap.Error(err)) continue } if !found { continue } endTs = checkpoint } region := bucket.Regions[0] timeRange, ok := timeRangeMap[region] if !ok { regionInfo := &domain.RegionInfo{Name: region, DisplayName: region} if r, ok := regions[region]; ok { regionInfo.DisplayName = r.DisplayName } timeRange = &domain.PiTRRegionTimeRange{Region: regionInfo} timeRangeMap[region] = timeRange } timeRange.TimeRanges = append(timeRange.TimeRanges, &domain.PiTRTimestampRange{ StartTS: timestamppb.New(time.Unix(logBackupTimeRange.StartTS, 0)), EndTS: timestamppb.New(time.Unix(endTs, 0)), }) } } var timeRanges []*domain.PiTRRegionTimeRange for _, timeRange := range timeRangeMap { timeRanges = append(timeRanges, timeRange) } return timeRanges, nil } ``` **Key differences from V1:** - Uses `bucket.Master` instead of `region == cluster.RegionName` to determine primary region - All logbackups treated uniformly (no "latest vs historical" split) - `cluster` and `cloudRegions` are passed in (pre-fetched by caller) - No pre-initialized `clusterRegionTimeRange` entry — map is populated organically - [x] **Step 2: Verify compile** ```bash cd repos/dataflow-service && go build ./app/... ``` Expected: no errors. --- ## Task 4: Create `pit_backup_testhelper.go` **Files:** - Create: `app/models/backup/impl/pit_backup_testhelper.go` External test packages (different directories) need to inject a mock storage function but cannot call unexported helpers from `impl`. The solution is a **non-test file** (regular `.go`, not `_test.go`) that exports `NewPitBackupModelWithStorageFunc`. > **Why not a `_test.go` file?** Go's test build constraints prevent `_test.go` files in one > package from being imported by test files in different packages/directories. - [x] **Step 1: Create `pit_backup_testhelper.go`** ```go package impl import ( "github.com/tidbcloud/dataflow-service/app/internal/feature_gate" "github.com/tidbcloud/dataflow-service/app/models/backup" baseRepo "github.com/tidbcloud/dataflow-service/app/repo" backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup" restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore" "github.com/tidbcloud/dataflow-service/app/workflow" "github.com/tidbcloud/dataflow-service/pkg/blob" "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli" "github.com/tidbcloud/pkgv2/db" ) // NewPitBackupModelWithStorageFunc creates a pitBackupModel with injectable getStorageFunc for testing. // IMPORTANT: This is test-only infrastructure but MUST be in a non-test file because // Go does not allow _test.go files to export symbols to external test packages in different directories. // External tests (e.g., backup_test/) need this function, so it must be in a regular .go file. // DO NOT remove or rename this file without also updating all external test packages that depend on it. func NewPitBackupModelWithStorageFunc( dbProvider db.TransProvider, pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo, logBackupRepo backupRepo.LogBackupRepo, backupRepo backupRepo.BackupRepo, restoreRepo restoreRepo.RestoreRepo, workflowCli workflow.Workflow, coreCli grpcli.CoreGrpcClient, snapshotBackupModel backup.SnapshotBackupModel, backupSettingModel backup.BackupSettingModel, scheduleRepo backupRepo.BackupScheduleRepo, featureGate feature_gate.FeatureGate, bucketRepo baseRepo.BucketRepo, getStorageFunc blob.GetStorageFunc, ) backup.PitBackupModel { m := NewPitBackupModel( dbProvider, pitBackupSwitchRepo, logBackupRepo, backupRepo, restoreRepo, bucketRepo, workflowCli, coreCli, snapshotBackupModel, backupSettingModel, scheduleRepo, featureGate, ) if getStorageFunc != nil { m.(*pitBackupModel).getStorageFunc = getStorageFunc } return m } ``` > **Note:** `NewPitBackupModelWithStorageFunc` delegates to `NewPitBackupModel` (passing > `bucketRepo` at position 6), then overrides `getStorageFunc` via type assertion. --- ## Task 5: Add unit tests in `pit_backup_test.go` **Files:** - Create: `app/models/backup/impl/pit_backup_test.go` Tests are in package `impl` (not external `backupv2_test`) to access `pitBackupModel` internals directly. The test file uses table-driven tests with helper functions. - [x] **Step 1: Write tests for `mergeSingleRegionPiTRTimeRanges`** Add tests covering: empty input, single range, touching/overlapping ranges, non-adjacent ranges, multiple ranges, equal start timestamps, newer start strictly after older end. - [x] **Step 2: Write tests for `groupPiTRTimeRangesByRegionV2`** Test cases (using injectable `checkpointFunc` via `pitBackupModel.getCheckpointForCRPiTR`): | Case | Setup | Expected | |------|-------|----------| | Empty logbackups | — | empty result | | Master bucket: region in cloud map | `Master: true` | uses cloud display name | | Master bucket: region absent from cloud map | `Master: true` | falls back to region name | | Non-master: checkpoint found | `Master: false`, func returns `(1800, true, nil)` | EndTS = 1800 | | Non-master: checkpoint not found | `Master: false`, func returns `(0, false, nil)` | bucket skipped | | Non-master: checkpoint error | `Master: false`, func returns error | bucket skipped (no propagation) | | Multiple logbackups: same region | 2× `Master: true` | 2 time ranges accumulated | | Master + non-master in same logbackup | mixed | 2 separate region entries | | `GetAnyClusterByID` error | cluster error | error propagated with "get cluster" annotation | | `ListCloudRegions` error | regions error | error propagated with "get cloud regions" annotation | **Helper functions used in tests:** ```go // ts converts Unix seconds to *timestamppb.Timestamp func ts(sec int64) *timestamppb.Timestamp // tsRange builds a PiTRTimestampRange from start/end Unix seconds func tsRange(start, end int64) *domain.PiTRTimestampRange // fakeCluster returns a minimal *core.Cluster func fakeCluster(projectID uint64, provider, region, regionDisplay string) *core.Cluster // fakeCloudRegions returns a cloud-regions map from name→displayName pairs func fakeCloudRegions(provider string, regions map[string]string) map[string]map[string]*core.CloudRegion // logBackupWithBuckets builds a PiTRTimeRange with the given buckets func logBackupWithBuckets(startTS, endTS int64, buckets []entity.BackupSettingBucket) *domain.PiTRTimeRange ``` - [x] **Step 3: Run tests** ```bash cd repos/dataflow-service && go test ./app/models/backup/impl/... -v 2>&1 | tail -40 ``` Expected: all tests pass. --- ## Task 6: Final integration check - [x] **Step 1: Build the entire app** ```bash cd repos/dataflow-service && go build ./app/... ``` Expected: no errors. - [x] **Step 2: Run the full app test suite** ```bash cd repos/dataflow-service && make app-unit-test-local ``` Expected: all tests pass (no regressions). --- ## Quick Reference **Key types:** - `domain.PiTRRegionTimeRange` — `{ Region *RegionInfo; TimeRanges []*PiTRTimestampRange }` - `domain.PiTRTimestampRange` — `{ StartTS, EndTS *timestamppb.Timestamp }` - `domain.PiTRTimeRange` — `{ LogBackup entity.LogBackup; StartTS, EndTS int64 }` - `entity.BackupSettingBucket.Master` — `bool`; `true` = primary region, `false` = secondary (CRPiTR) - `entity.BucketSetting.GetBucketByRegion(region string) (BackupSettingBucket, bool)` — in `bucket_ext.go` **Key functions:** - `util.TransToTS(tso uint64) int64` — converts raw TSO to physical seconds - `domain.BuildDataflowServiceInfraRoleARN(awsAccountID string) string` — constructs AWS Role ARN - `blob.NewStorage` — default `GetStorageFunc` (note: original has typo `blob.DefualtGetStorageFunc` elsewhere in codebase) - `blob.IsNotFoundError(err error) bool` — checks if error is "not found" - `BuildStorageOption(bucket *entity.Bucket, region string) *blob.StorageOption` — in `pit_backup_checkpoint.go` **Test run command:** ```bash cd repos/dataflow-service && go test ./app/models/backup/impl/... -v ``` **Pending follow-on work:** - Wire `groupPiTRTimeRangesByRegionV2` into `mergePiTRTimeRanges` (replace V1 call), possibly behind a feature gate