# CRRF: Record LogBackupID in CRRFSetting — 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:** Persist the log backup ID into `CRRFSetting` when CRRF is enabled, so the disable workflow reliably references the exact log backup that CRRF was set up with, even after PiTR toggles. **Architecture:** Add `LogBackupID` to `CRRFSetting` and a DB migration. In `enable_pitr.RunAfter`, after confirming the running log backup, save its ID to `CRRFSetting` (skipping if already matches). In `LoadCRRFActionContext`, prefer the stored ID over querying for a running log backup. Clear `LogBackupID` in `Disable()`. **Tech Stack:** Go, GORM, testify/mock, testify/suite --- ## File Map | File | Action | Responsibility | |------|--------|---------------| | `app/repo/entity/crrf.go` | Modify | Add `LogBackupID` field | | `app/repo/entity/crrf_ext.go` | Modify | Clear `LogBackupID` in `Disable()` | | `migrate/record_202605201000_add_log_backup_id_to_crrf_setting.go` | Create | Add `log_backup_id` column via AutoMigrate | | `app/workflow/action/crrf/enable/enable_pitr.go` | Modify | Add `crrfSettingRepo` dep; save `LogBackupID` in `RunAfter` | | `app/workflow/action/crrf/enable/enable_pitr_test.go` | Modify | Add `crrfSettingRepo` mock; new `RunAfter` test cases | | `app/workflow/action/crrf/common/load_crrf_action_context.go` | Modify | Prefer `LogBackupID` via `Get+IDEq` over querying running backup | | `app/workflow/action/crrf/common/load_crrf_action_context_test.go` | Create | Test both ID-based and fallback paths | | `app/workflow/dag/crrf/enable.go` | Modify | Pass `c.CRRFSettingRepo` to `NewEnablePitrAction` | --- ## Task 1: Schema — add `LogBackupID` field, update `Disable()`, add migration **Files:** - Modify: `app/repo/entity/crrf.go` - Modify: `app/repo/entity/crrf_ext.go` - Create: `migrate/record_202605201000_add_log_backup_id_to_crrf_setting.go` - [ ] **Step 1: Add `LogBackupID` to `CRRFSetting`** In `app/repo/entity/crrf.go`, insert after the `SecondaryBucketID` line: ```go // before (line 28 area): MasterBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` SecondaryBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` // Error tracking for enable workflow failures. ErrMsg *string `gorm:"type:text"` // after: MasterBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` SecondaryBucketID ID `gorm:"type:bigint(20) unsigned;default:0"` // LogBackupID is the ID of the log backup used when CRRF was enabled. // Recorded in enable_pitr.RunAfter and cleared on disable. LogBackupID ID `gorm:"type:bigint(20) unsigned;default:0"` // Error tracking for enable workflow failures. ErrMsg *string `gorm:"type:text"` ``` - [ ] **Step 2: Clear `LogBackupID` in `Disable()`** In `app/repo/entity/crrf_ext.go`, add `s.LogBackupID = 0` after the other field clears: ```go func (s *CRRFSetting) Disable(ctx context.Context) (changed bool, err error) { logger := log.GetLogger(ctx) if s.Status == CRRFSettingStatusDisabled { logger.Info("crrf setting already disabled, idempotent skip") return false, nil } if s.Status != CRRFSettingStatusDisabling { return false, fmt.Errorf("crrf setting status is not disabling, cannot disable, current status: %s", s.Status) } s.Status = CRRFSettingStatusDisabled s.SecondaryRegion = "" s.MasterBucketID = 0 s.SecondaryBucketID = 0 s.LogBackupID = 0 s.ErrMsg = nil return true, nil } ``` - [ ] **Step 3: Create the migration file** Create `migrate/record_202605201000_add_log_backup_id_to_crrf_setting.go`: ```go package migrate import ( "github.com/pingcap/errors" "github.com/pingcap/log" "github.com/tidbcloud/dataflow-service/app/repo/entity" "go.uber.org/zap" "gorm.io/gorm" ) var _ = AddDataflowDBMigrateRecord( "202605201000", func(db *gorm.DB) error { if err := db.AutoMigrate(&entity.CRRFSetting{}); err != nil { errMsg := "auto migrate 'crrf_settings' failed" log.Error(errMsg, zap.Error(err)) return errors.Annotate(err, errMsg) } log.Info("migration 202605201000 completed: add log_backup_id to crrf_settings") return nil }, ) ``` - [ ] **Step 4: Build to verify compilation** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go build ./app/repo/entity/... ./migrate/... ``` Expected: no errors. - [ ] **Step 5: Commit** ```bash git add app/repo/entity/crrf.go app/repo/entity/crrf_ext.go \ migrate/record_202605201000_add_log_backup_id_to_crrf_setting.go git commit -m "feat(crrf): add LogBackupID field to CRRFSetting and migration" ``` --- ## Task 2: Record `LogBackupID` in `enable_pitr.RunAfter` **Files:** - Modify: `app/workflow/action/crrf/enable/enable_pitr.go` - Modify: `app/workflow/action/crrf/enable/enable_pitr_test.go` - Modify: `app/workflow/dag/crrf/enable.go` - [ ] **Step 1: Write failing tests** Replace the entire `app/workflow/action/crrf/enable/enable_pitr_test.go` with: ```go package enable import ( "context" "testing" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/suite" backupModelMock "github.com/tidbcloud/dataflow-service/app/models/backup/mock" "github.com/tidbcloud/dataflow-service/app/repo/entity" repoMock "github.com/tidbcloud/dataflow-service/app/repo/mock" "github.com/tidbcloud/dataflow-service/app/workflow/action" "github.com/tidbcloud/pkgv2/bizerr" "go.uber.org/zap" ) type enablePitrTestSuite struct { suite.Suite act *enablePitrAction backupSettingRepo *repoMock.BackupSettingRepo backupSettingModel *backupModelMock.BackupSettingModel logBackupRepo *repoMock.LogBackupRepo crrfSettingRepo *repoMock.CRRFSettingRepo logger *zap.Logger } func TestEnablePitrSuite(t *testing.T) { suite.Run(t, new(enablePitrTestSuite)) } func (s *enablePitrTestSuite) SetupSuite() { s.backupSettingRepo = repoMock.NewBackupSettingRepo(s.T()) s.backupSettingModel = backupModelMock.NewBackupSettingModel(s.T()) s.logBackupRepo = repoMock.NewLogBackupRepo(s.T()) s.crrfSettingRepo = repoMock.NewCRRFSettingRepo(s.T()) s.logger = zap.NewNop() s.act = &enablePitrAction{ backupSettingRepo: s.backupSettingRepo, backupSettingModel: s.backupSettingModel, logBackupRepo: s.logBackupRepo, crrfSettingRepo: s.crrfSettingRepo, } } func (s *enablePitrTestSuite) TearDownTest() { s.backupSettingRepo.AssertExpectations(s.T()) s.backupSettingModel.AssertExpectations(s.T()) s.logBackupRepo.AssertExpectations(s.T()) s.crrfSettingRepo.AssertExpectations(s.T()) } func (s *enablePitrTestSuite) params() *enablePitrParams { return &enablePitrParams{ClusterID: 1001} } // --- Run tests --- func (s *enablePitrTestSuite) TestRun() { ctx := context.Background() p := s.params() tests := map[string]struct { setup func() wantErr bool errIs error }{ "repo error": { setup: func() { s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything). Return(nil, bizerr.New(bizerr.ErrCodeDB, "db error")).Once() }, wantErr: true, }, "backup setting not found": { setup: func() { s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(nil, nil).Once() }, wantErr: true, errIs: action.StopRetry, }, "PiTR already enabled — idempotent skip": { setup: func() { bs := &entity.BackupSetting{PiTR: &entity.PiTRSetting{Enable: true}} s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once() }, wantErr: false, }, "trigger succeeds": { setup: func() { bs := &entity.BackupSetting{PiTR: &entity.PiTRSetting{Enable: false}} s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once() s.backupSettingModel.On("ModifyPitBackupSetting", mock.Anything, mock.Anything).Return(nil).Once() }, wantErr: false, }, "trigger fails": { setup: func() { bs := &entity.BackupSetting{PiTR: &entity.PiTRSetting{Enable: false}} s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once() s.backupSettingModel.On("ModifyPitBackupSetting", mock.Anything, mock.Anything). Return(bizerr.New(bizerr.ErrCodeUnknown, "remote error")).Once() }, wantErr: true, }, "PiTR field nil — triggers enable": { setup: func() { bs := &entity.BackupSetting{PiTR: nil} s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once() s.backupSettingModel.On("ModifyPitBackupSetting", mock.Anything, mock.Anything).Return(nil).Once() }, wantErr: false, }, } for name, tc := range tests { s.Run(name, func() { tc.setup() err := s.act.Run(ctx, s.logger, p) if tc.wantErr { s.Error(err) if tc.errIs != nil { s.Equal(tc.errIs, err) } } else { s.NoError(err) } }) } } // --- RunAfter tests --- func (s *enablePitrTestSuite) TestRunAfter() { ctx := context.Background() p := s.params() tests := map[string]struct { setup func() wantErr bool }{ "list fails": { setup: func() { s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return(nil, bizerr.New(bizerr.ErrCodeDB, "db error")).Once() }, wantErr: true, }, "no running log backup yet": { setup: func() { s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{}, nil).Once() }, wantErr: true, }, "GetByClusterID fails after log backup found": { setup: func() { lb := makeRunningLogBackup(p.ClusterID, 99) s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{lb}, nil).Once() s.crrfSettingRepo.On("GetByClusterID", mock.Anything, p.ClusterID). Return(nil, bizerr.New(bizerr.ErrCodeDB, "db error")).Once() }, wantErr: true, }, "log backup ID already matches — idempotent skip, no save": { setup: func() { lb := makeRunningLogBackup(p.ClusterID, 99) crrf := &entity.CRRFSetting{LogBackupID: entity.ID(99)} s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{lb}, nil).Once() s.crrfSettingRepo.On("GetByClusterID", mock.Anything, p.ClusterID). Return(crrf, nil).Once() // No Save call expected. }, wantErr: false, }, "log backup ID differs — updates CRRFSetting": { setup: func() { lb := makeRunningLogBackup(p.ClusterID, 99) crrf := &entity.CRRFSetting{LogBackupID: entity.ID(77)} // stale / first enable s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{lb}, nil).Once() s.crrfSettingRepo.On("GetByClusterID", mock.Anything, p.ClusterID). Return(crrf, nil).Once() s.crrfSettingRepo.On("Save", mock.Anything, crrf).Return(nil).Once() }, wantErr: false, }, "save fails": { setup: func() { lb := makeRunningLogBackup(p.ClusterID, 99) crrf := &entity.CRRFSetting{LogBackupID: entity.ID(0)} s.logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{lb}, nil).Once() s.crrfSettingRepo.On("GetByClusterID", mock.Anything, p.ClusterID). Return(crrf, nil).Once() s.crrfSettingRepo.On("Save", mock.Anything, crrf). Return(bizerr.New(bizerr.ErrCodeDB, "save failed")).Once() }, wantErr: true, }, } for name, tc := range tests { s.Run(name, func() { tc.setup() err := s.act.RunAfter(ctx, s.logger, p) if tc.wantErr { s.Error(err) } else { s.NoError(err) } }) } } func makeRunningLogBackup(clusterID uint64, id uint64) entity.LogBackup { lb := entity.LogBackup{Status: entity.LogBackupStatusRunning} lb.ID = entity.ID(id) lb.ClusterID = entity.ID(clusterID) return lb } ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/workflow/action/crrf/enable/... -run TestEnablePitrSuite -v 2>&1 | tail -20 ``` Expected: compilation error — `enablePitrAction` has no field `crrfSettingRepo`. - [ ] **Step 3: Update `enable_pitr.go`** Replace the file content with: ```go package enable import ( "context" "fmt" "time" "github.com/avast/retry-go/v4" backupModel "github.com/tidbcloud/dataflow-service/app/models/backup" "github.com/tidbcloud/dataflow-service/app/models/domain" backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup" crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf" "github.com/tidbcloud/dataflow-service/app/repo/entity" "github.com/tidbcloud/dataflow-service/app/workflow" "github.com/tidbcloud/dataflow-service/app/workflow/action" "go.uber.org/zap" ) type enablePitrParams struct { ClusterID uint64 `json:"cluster_id"` } type enablePitrAction struct { backupSettingRepo backupRepo.BackupSettingRepo backupSettingModel backupModel.BackupSettingModel logBackupRepo backupRepo.LogBackupRepo crrfSettingRepo crrfRepo.CRRFSettingRepo } // NewEnablePitrAction returns a Wrapper that triggers PiTR log backup in Run // and polls until a running log backup is found in RunAfter. func NewEnablePitrAction( backupSettingRepo backupRepo.BackupSettingRepo, backupSettingModel backupModel.BackupSettingModel, logBackupRepo backupRepo.LogBackupRepo, crrfSettingRepo crrfRepo.CRRFSettingRepo, ) *action.Wrapper { return action.MustNewActionWrapper( workflow.ActionCRRFEnablePitr, &enablePitrAction{ backupSettingRepo: backupSettingRepo, backupSettingModel: backupSettingModel, logBackupRepo: logBackupRepo, crrfSettingRepo: crrfSettingRepo, }, action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry), ) } // invokeRunAfterWithSlowExponentialRetry uses exponential backoff starting at 15s up to 5min, // suitable for long-running waits like PiTR enable which may take 24h+ (waiting for scheduled snapshot backup). // //nolint:gomnd func invokeRunAfterWithSlowExponentialRetry(ctx context.Context, logger *zap.Logger, run func() error) error { return retry.Do(func() error { return run() }, retry.Context(ctx), retry.DelayType(retry.BackOffDelay), retry.Delay(15*time.Second), retry.MaxDelay(5*time.Minute), retry.Attempts(0), retry.LastErrorOnly(true), retry.OnRetry(func(n uint, err error) { logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err)) }), retry.RetryIf(func(err error) bool { return err != action.StopRetry }), ) } func (a *enablePitrAction) Params() interface{} { return &enablePitrParams{} } // Run triggers PiTR log backup. It is idempotent: if PiTR is already enabled, exits immediately. // The Wrapper calls this with finite exponential retry. func (a *enablePitrAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error { p := params.(*enablePitrParams) backupSetting, bizErr := a.backupSettingRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{ ClusterID: &p.ClusterID, }) if bizErr != nil { return bizErr } if backupSetting == nil { logger.Error("backup setting not found", zap.Uint64("clusterID", p.ClusterID)) return action.StopRetry } // Already enabled — nothing to trigger. if backupSetting.PiTR != nil && backupSetting.PiTR.Enable { logger.Info("PiTR already enabled, skipping trigger", zap.Uint64("clusterID", p.ClusterID)) return nil } if bizErr = a.backupSettingModel.ModifyPitBackupSetting(ctx, &domain.PitSettingModificationReq{ ClusterID: p.ClusterID, Enable: true, }); bizErr != nil { logger.Warn("failed to trigger PiTR enable", zap.Error(bizErr)) return bizErr } logger.Info("PiTR enable triggered", zap.Uint64("clusterID", p.ClusterID)) return nil } // RunAfter polls until a running log backup is found, then records its ID in CRRFSetting. // The Wrapper calls this with infinite retry until it succeeds or the task times out. func (a *enablePitrAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error { p := params.(*enablePitrParams) logBackups, bizErr := a.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet { return qs.ClusterIDEq(entity.ID(p.ClusterID)).StatusEq(entity.LogBackupStatusRunning) }) if bizErr != nil { return bizErr } if len(logBackups) == 0 { return fmt.Errorf("PiTR log backup not yet running for cluster %d", p.ClusterID) } lb := logBackups[0] logger.Info("PiTR ready: running log backup found", zap.Uint64("clusterID", p.ClusterID), zap.Uint64("logBackupID", uint64(lb.ID))) setting, bizErr := a.crrfSettingRepo.GetByClusterID(ctx, p.ClusterID) if bizErr != nil { return bizErr } if setting.LogBackupID != lb.ID { setting.LogBackupID = lb.ID if saveErr := a.crrfSettingRepo.Save(ctx, setting); saveErr != nil { return saveErr } logger.Info("CRRFSetting.LogBackupID recorded", zap.Uint64("clusterID", p.ClusterID), zap.Uint64("logBackupID", uint64(lb.ID))) } return nil } ``` - [ ] **Step 4: Update `dag/crrf/enable.go` to pass `crrfSettingRepo`** In `app/workflow/dag/crrf/enable.go`, change line 28 from: ```go crrf_enable.NewEnablePitrAction(c.BackupSettingRepo, c.BackupSettingModel, c.LogBackupRepo), ``` to: ```go crrf_enable.NewEnablePitrAction(c.BackupSettingRepo, c.BackupSettingModel, c.LogBackupRepo, c.CRRFSettingRepo), ``` - [ ] **Step 5: Run tests** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/workflow/action/crrf/enable/... -run TestEnablePitrSuite -v 2>&1 | tail -30 ``` Expected: all subtests pass. - [ ] **Step 6: Build to verify no compilation errors** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go build ./... ``` Expected: no errors. - [ ] **Step 7: Commit** ```bash git add app/workflow/action/crrf/enable/enable_pitr.go \ app/workflow/action/crrf/enable/enable_pitr_test.go \ app/workflow/dag/crrf/enable.go git commit -m "feat(crrf): record LogBackupID in CRRFSetting during enable_pitr.RunAfter" ``` --- ## Task 3: Use stored `LogBackupID` in `LoadCRRFActionContext` **Files:** - Modify: `app/workflow/action/crrf/common/load_crrf_action_context.go` - Create: `app/workflow/action/crrf/common/load_crrf_action_context_test.go` - [ ] **Step 1: Write failing tests** Create `app/workflow/action/crrf/common/load_crrf_action_context_test.go`: ```go package common import ( "context" "testing" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup" crrfRepoMock "github.com/tidbcloud/dataflow-service/app/repo/crrf/mock" "github.com/tidbcloud/dataflow-service/app/repo/entity" repoMock "github.com/tidbcloud/dataflow-service/app/repo/mock" "github.com/tidbcloud/pkgv2/bizerr" ) // ---- helpers ----------------------------------------------------------------- func makeCtxBackupSetting(clusterID uint64) *entity.BackupSetting { return &entity.BackupSetting{ Updatable: entity.Updatable{Base: entity.Base{ID: 1}}, TenantID: entity.ID(100), InfraProviderType: "aws", Bucket: &entity.BucketSetting{ Buckets: []entity.BackupSettingBucket{ {Master: true, Regions: []string{"us-east-1"}, BucketId: 1}, }, }, } } func makeCtxCRRFSetting(clusterID uint64, logBackupID uint64) *entity.CRRFSetting { return &entity.CRRFSetting{ Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}}, ClusterID: entity.ID(clusterID), LogBackupID: entity.ID(logBackupID), } } func makeLogBackupWithSchedule(id uint64, scheduleName string) *entity.LogBackup { lb := &entity.LogBackup{ BackupSchedule: &entity.BackupSchedule{Name: scheduleName}, } lb.ID = entity.ID(id) return lb } // ---- LoadCRRFActionContext tests --------------------------------------------- func TestLoadCRRFActionContext_LogBackupIDStored(t *testing.T) { // When CRRFSetting.LogBackupID is non-zero, LoadCRRFActionContext should use // logBackupRepo.Get with IDEq+PreloadBackupSchedule instead of querying running backup. clusterID := uint64(42) logBackupID := uint64(77) crrfRepo := crrfRepoMock.NewCRRFSettingRepo(t) backupSettingRepo := repoMock.NewBackupSettingRepo(t) logBackupRepo := repoMock.NewLogBackupRepo(t) crrfRepo.On("GetByClusterID", mock.Anything, clusterID). Return(makeCtxCRRFSetting(clusterID, logBackupID), nil).Once() backupSettingRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}). Return(makeCtxBackupSetting(clusterID), nil).Once() logBackupRepo.On("Get", mock.Anything, mock.Anything). Return(makeLogBackupWithSchedule(logBackupID, "sched"), nil).Once() ctx, err := LoadCRRFActionContext(ctx(t), clusterID, backupSettingRepo, crrfRepo, logBackupRepo) require.NoError(t, err) require.NotNil(t, ctx) require.Equal(t, "log-sched-77", ctx.LogBackupName) // List must NOT have been called (Get was used instead). logBackupRepo.AssertNotCalled(t, "List") } func TestLoadCRRFActionContext_LogBackupIDZero_FallsBackToRunning(t *testing.T) { // When CRRFSetting.LogBackupID == 0, fall back to FindRunningLogBackupCRName. clusterID := uint64(42) crrfRepo := crrfRepoMock.NewCRRFSettingRepo(t) backupSettingRepo := repoMock.NewBackupSettingRepo(t) logBackupRepo := repoMock.NewLogBackupRepo(t) crrfRepo.On("GetByClusterID", mock.Anything, clusterID). Return(makeCtxCRRFSetting(clusterID, 0), nil).Once() backupSettingRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}). Return(makeCtxBackupSetting(clusterID), nil).Once() logBackupRepo.On("List", mock.Anything, mock.Anything). Return([]entity.LogBackup{makeLogBackup(clusterID, "lb-001", entity.LogBackupStatusRunning)}, nil).Once() actCtx, err := LoadCRRFActionContext(ctx(t), clusterID, backupSettingRepo, crrfRepo, logBackupRepo) require.NoError(t, err) require.NotNil(t, actCtx) require.Equal(t, "log-schedule-1", actCtx.LogBackupName) // Get must NOT have been called. logBackupRepo.AssertNotCalled(t, "Get") } func TestLoadCRRFActionContext_GetByIDFails(t *testing.T) { clusterID := uint64(42) logBackupID := uint64(77) crrfRepo := crrfRepoMock.NewCRRFSettingRepo(t) backupSettingRepo := repoMock.NewBackupSettingRepo(t) logBackupRepo := repoMock.NewLogBackupRepo(t) crrfRepo.On("GetByClusterID", mock.Anything, clusterID). Return(makeCtxCRRFSetting(clusterID, logBackupID), nil).Once() backupSettingRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}). Return(makeCtxBackupSetting(clusterID), nil).Once() logBackupRepo.On("Get", mock.Anything, mock.Anything). Return(nil, bizerr.New(bizerr.ErrCodeDB, "db error")).Once() _, err := LoadCRRFActionContext(ctx(t), clusterID, backupSettingRepo, crrfRepo, logBackupRepo) require.Error(t, err) } func ctx(t *testing.T) context.Context { t.Helper() return context.Background() } ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/workflow/action/crrf/common/... -run TestLoadCRRFActionContext -v 2>&1 | tail -20 ``` Expected: `TestLoadCRRFActionContext_LogBackupIDStored` and `TestLoadCRRFActionContext_GetByIDFails` fail because `LoadCRRFActionContext` still always calls `List`. - [ ] **Step 3: Update `LoadCRRFActionContext`** Replace `load_crrf_action_context.go` with: ```go package common import ( "context" "fmt" "github.com/tidbcloud/dataflow-service/app/models/domain" backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup" crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf" "github.com/tidbcloud/dataflow-service/app/repo/entity" "github.com/tidbcloud/dataflow-service/app/workflow/action" regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api" ) // CRRFActionContext holds the context resolved by LoadCRRFActionContext and // shared across CRRF workflow actions (enable, disable, etc.). type CRRFActionContext struct { MasterRegion *regionalapi.Region CRRFSetting *entity.CRRFSetting BackupSetting *entity.BackupSetting LogBackupName string } // FindRunningLogBackupCRName returns the name of the first running log backup for the cluster. // It returns an error if no running log backup is found. func FindRunningLogBackupCRName(ctx context.Context, clusterID uint64, repo backupRepo.LogBackupRepo) (string, error) { logBackups, bizErr := repo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet { return qs.ClusterIDEq(entity.ID(clusterID)).StatusEq(entity.LogBackupStatusRunning).PreloadBackupSchedule() }) if bizErr != nil { return "", bizErr } if len(logBackups) == 0 { return "", fmt.Errorf("no running log backup found for cluster %d", clusterID) } logBackup := logBackups[0] return logBackup.CRName(), nil } // LoadCRRFActionContext resolves the shared context needed by CRRF workflow actions: // the CRRFSetting, the regional endpoint, and the log backup name. // // When CRRFSetting.LogBackupID is non-zero (set during enable), the log backup is fetched // by ID to ensure the disable workflow references the exact backup that CRRF was set up with, // even if PiTR was toggled (which would create a new running log backup). // When LogBackupID is zero (old records without the field), falls back to finding a running one. func LoadCRRFActionContext( ctx context.Context, clusterID uint64, backupSettingRepo backupRepo.BackupSettingRepo, crrfSettingRepo crrfRepo.CRRFSettingRepo, logBackupRepo backupRepo.LogBackupRepo, ) (*CRRFActionContext, error) { crrfSetting, bizErr := crrfSettingRepo.GetByClusterID(ctx, clusterID) if bizErr != nil { return nil, bizErr } backupSetting, bizErr := backupSettingRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{ ClusterID: &clusterID, }) if bizErr != nil { return nil, bizErr } if backupSetting == nil { return nil, action.StopRetry } masterRegion, ok := backupSetting.MasterRegion() if !ok { return nil, fmt.Errorf("master region not found for cluster %d", clusterID) } var logBackupName string if crrfSetting.LogBackupID != 0 { lb, bizErr := logBackupRepo.Get(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet { return qs.IDEq(crrfSetting.LogBackupID).PreloadBackupSchedule() }) if bizErr != nil { return nil, bizErr } if lb == nil { return nil, fmt.Errorf("log backup %d not found for cluster %d", crrfSetting.LogBackupID, clusterID) } logBackupName = lb.CRName() } else { var err error logBackupName, err = FindRunningLogBackupCRName(ctx, clusterID, logBackupRepo) if err != nil { return nil, err } } return &CRRFActionContext{ MasterRegion: ®ionalapi.Region{ TenantID: uint64(backupSetting.TenantID), Name: masterRegion, CloudProvider: string(domain.InfraProviderType2CloudProviderType(backupSetting.InfraProviderType)), }, CRRFSetting: crrfSetting, BackupSetting: backupSetting, LogBackupName: logBackupName, }, nil } ``` - [ ] **Step 4: Run all common package tests** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/workflow/action/crrf/common/... -v 2>&1 | tail -30 ``` Expected: all tests pass, including the three new `TestLoadCRRFActionContext_*` tests. - [ ] **Step 5: Run all CRRF action tests to verify no regressions** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/workflow/action/crrf/... -v 2>&1 | grep -E "^(--- |PASS|FAIL|ok)" | tail -40 ``` Expected: all tests pass. - [ ] **Step 6: Build** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go build ./... ``` Expected: no errors. - [ ] **Step 7: Commit** ```bash git add app/workflow/action/crrf/common/load_crrf_action_context.go \ app/workflow/action/crrf/common/load_crrf_action_context_test.go git commit -m "feat(crrf): use stored LogBackupID in LoadCRRFActionContext for reliable disable" ``` --- ## Final Verification - [ ] **Run the full unit test suite** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service && go test ./app/... -count=1 2>&1 | grep -E "^(ok|FAIL|---)" | tail -30 ``` Expected: all packages pass, no FAILs.