Skip to content

syncer(dm): prevent checkpoint flush after BeginTx error #12644

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
GMHDBJD wants to merge 1 commit into
base: master
Choose a base branch
from
Open

syncer(dm): prevent checkpoint flush after BeginTx error #12644

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions dm/syncer/checkpoint_flush_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import (
"github.com/pingcap/failpoint"
"github.com/pingcap/tidb/pkg/util/filter"
tcontext "github.com/pingcap/tiflow/dm/pkg/context"
"github.com/pingcap/tiflow/dm/pkg/terror"
"go.uber.org/atomic"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -77,7 +76,7 @@ func (w *checkpointFlushWorker) Run(ctx *tcontext.Context) {
// optimistic shard info, DM-master may resolved the optimistic lock and let other worker execute DDL. So after this
// worker resume, it can not execute the DML/DDL in old binlog because of downstream table structure mismatching.
// We should find a way to (compensating) implement a transaction containing interaction with both etcd and SQL.
if err != nil && (terror.ErrDBExecuteFailed.Equal(err) || terror.ErrDBUnExpect.Equal(err)) {
if isDownstreamExecutionError(err) {
ctx.L().Warn(fmt.Sprintf("error detected when executing SQL job, skip %s checkpoint and shutdown checkpointFlushWorker", flushLogMsg),
zap.Stringer("globalPos", task.snapshotInfo.globalPos),
zap.Error(err))
Expand Down
172 changes: 172 additions & 0 deletions dm/syncer/checkpoint_flush_worker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
// Copyright 2026 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package syncer

import (
"context"
"database/sql"
"testing"
"time"

"github.com/pingcap/tidb/pkg/meta/model"
"github.com/pingcap/tidb/pkg/util/filter"
"github.com/pingcap/tiflow/dm/pkg/binlog"
tcontext "github.com/pingcap/tiflow/dm/pkg/context"
"github.com/pingcap/tiflow/dm/pkg/schema"
"github.com/pingcap/tiflow/dm/pkg/terror"
"github.com/stretchr/testify/require"
"go.uber.org/atomic"
)

func TestCheckpointFlushWorkerSkipsCheckpointOnBeginError(t *testing.T) {
t.Parallel()

cp := &checkpointFlushWorkerTestCheckpoint{}
var execError atomic.Error
execError.Store(terror.ErrDBExecuteFailedBegin.Delegate(sql.ErrConnDone))

worker := &checkpointFlushWorker{
input: make(chan *checkpointFlushTask, 1),
cp: cp,
execError: &execError,
afterFlushFn: func(*checkpointFlushTask) error { return nil },
updateJobMetricsFn: func(bool, string, *job) {},
}

done := make(chan struct{})
go func() {
worker.Run(tcontext.Background())
close(done)
}()

errCh := make(chan error, 1)
worker.Add(&checkpointFlushTask{
snapshotInfo: &SnapshotInfo{id: 1},
syncFlushErrCh: errCh,
})
require.NoError(t, <-errCh)

worker.Close()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("checkpoint flush worker did not stop")
}
require.Zero(t, cp.flushCount, "checkpoint flush must be skipped after downstream BeginTx failure; flushing here can persist a checkpoint past non-durable DML")
}

type checkpointFlushWorkerTestCheckpoint struct {

@ingress-bot ingress-bot May 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 [Minor] New test hand-rolls full CheckPoint mock instead of reusing the package's interface-embedding pattern

Impact
Adds about 100 lines of boilerplate and a third divergent CheckPoint mock in the same package.
This one must be manually edited every time the CheckPoint interface grows a method, while the embedded-interface mocks do not, creating ongoing maintenance drift and compile-break risk for unrelated changes.

Scope

  • dm/syncer/checkpoint_flush_worker_test.go:70 — checkpointFlushWorkerTestCheckpoint
  • dm/syncer/status_test.go:72 — mockCheckpoint
  • dm/syncer/safe_mode_test.go:30 — mockCheckpointForSafeMode

Evidence
mockCheckpoint and mockCheckpointForSafeMode embed the CheckPoint interface and override only the methods they exercise. checkpointFlushWorkerTestCheckpoint instead implements every CheckPoint method explicitly, yet the test's worker returns early via isDownstreamExecutionError so FlushPointsExcept (and every other method) is never actually called.

Change request
Follow the existing pattern: declare type checkpointFlushWorkerTestCheckpoint struct { CheckPoint; flushCount int } and keep only the FlushPointsExcept override that increments flushCount, dropping the ~25 unused method stubs.

flushCount int
}

func (c *checkpointFlushWorkerTestCheckpoint) Init(*tcontext.Context) error { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) Close() {}

func (c *checkpointFlushWorkerTestCheckpoint) ResetConn(*tcontext.Context) error { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) Clear(*tcontext.Context) error { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) Load(*tcontext.Context) error { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) LoadMeta(context.Context) error { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) SaveTablePoint(*filter.Table, binlog.Location, *model.TableInfo) {
}

func (c *checkpointFlushWorkerTestCheckpoint) DeleteTablePoint(*tcontext.Context, *filter.Table) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) DeleteAllTablePoint(*tcontext.Context) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) DeleteSchemaPoint(*tcontext.Context, string) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) IsOlderThanTablePoint(*filter.Table, binlog.Location) bool {
return false
}

func (c *checkpointFlushWorkerTestCheckpoint) SaveGlobalPoint(binlog.Location) {}

func (c *checkpointFlushWorkerTestCheckpoint) SaveGlobalPointForcibly(binlog.Location) {}

func (c *checkpointFlushWorkerTestCheckpoint) Snapshot(bool) *SnapshotInfo { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) DiscardPendingSnapshots() {}

func (c *checkpointFlushWorkerTestCheckpoint) FlushPointsExcept(
*tcontext.Context,
int,
[]*filter.Table,
[]string,
[][]interface{},
) error {
c.flushCount++
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) FlushPointsWithTableInfos(
*tcontext.Context,
[]*filter.Table,
[]*model.TableInfo,
) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) FlushSafeModeExitPoint(*tcontext.Context) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) GlobalPoint() binlog.Location { return binlog.Location{} }

func (c *checkpointFlushWorkerTestCheckpoint) GlobalPointSaveTime() time.Time { return time.Time{} }

func (c *checkpointFlushWorkerTestCheckpoint) SaveSafeModeExitPoint(*binlog.Location) {}

func (c *checkpointFlushWorkerTestCheckpoint) SafeModeExitPoint() *binlog.Location { return nil }

func (c *checkpointFlushWorkerTestCheckpoint) TablePoint() map[string]map[string]binlog.Location {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) GetTableInfo(string, string) *model.TableInfo {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) FlushedGlobalPoint() binlog.Location {
return binlog.Location{}
}

func (c *checkpointFlushWorkerTestCheckpoint) LastFlushOutdated() bool { return false }

func (c *checkpointFlushWorkerTestCheckpoint) Rollback() {}

func (c *checkpointFlushWorkerTestCheckpoint) String() string { return "" }

func (c *checkpointFlushWorkerTestCheckpoint) LoadIntoSchemaTracker(context.Context, *schema.Tracker) error {
return nil
}

func (c *checkpointFlushWorkerTestCheckpoint) CheckAndUpdate(
context.Context,
map[string]string,
map[string]map[string]string,
) error {
return nil
}
2 changes: 1 addition & 1 deletion dm/syncer/dml_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,7 +365,7 @@ func (w *DMLWorker) executeBatchJobs(queueID int, jobs []*job, disableForeignKey
}
})

if w.judgeKeyNotFound(affect, jobs) {
if err == nil && w.judgeKeyNotFound(affect, jobs) {
// throw an error if needed in the future.
// err = terror.ErrDBExecuteFailed.Delegate(errors.New("key not found"), "mock")
w.logger.Warn("no matching record is found to update/delete, ER_KEY_NOT_FOUND", zap.Int("affect", affect), zap.Int("jobs", len(jobs)), zap.Stringer("start from", jobs[0].startLocation), zap.Stringer("end at", jobs[len(jobs)-1].currentLocation))
Expand Down
24 changes: 24 additions & 0 deletions dm/syncer/execution_error.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
// Copyright 2026 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package syncer

import "github.com/pingcap/tiflow/dm/pkg/terror"

func isDownstreamExecutionError(err error) bool {

@ingress-bot ingress-bot May 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

2 findings on this line:

🟡 [Minor] Resume-safety predicate is only partially covered: no negative case and three of four call sites untested

Impact
The new test asserts checkpoint flush is skipped for one error class (ErrDBExecuteFailedBegin) at one of four call sites, and never asserts that a non-downstream error still flushes.
A degenerate predicate that returned true for every non-nil error (e.g. wrongly matching context.Canceled) would still pass this test, yet would silently stall checkpoint progress; conversely, dropping ErrDBExecuteFailed/ErrDBUnExpect or breaking the inverted safe-mode-exit branch at syncer.go:1977 would reintroduce the checkpoint-advances-past-uncommitted-DML data loss on resume without failing CI.

Scope

  • dm/syncer/execution_error.go:19 — isDownstreamExecutionError
  • dm/syncer/checkpoint_flush_worker_test.go:60 — TestCheckpointFlushWorkerSkipsCheckpointOnBeginError
  • dm/syncer/syncer.go:1977 — Syncer.Run

Evidence
isDownstreamExecutionError guards four resume-critical branches with inverted meaning: it skips checkpoint flush in checkpointFlushWorker.Run, flushCheckPoints, and flushCheckPointsAsync, but triggers FlushSafeModeExitPoint at syncer.go:1977. The single test only exercises the worker skip path with ErrDBExecuteFailedBegin; ErrDBExecuteFailed, ErrDBUnExpect, the negative (non-matching) case, and the safe-mode-exit branch are unverified.

Change request
Add a focused unit test for isDownstreamExecutionError covering all three positive error classes plus at least one negative (e.g. a generic/context.Canceled error returning false), so the predicate contract that all four sites depend on is locked down independent of any one call site.

🟡 [Minor] isDownstreamExecutionError name over-promises generality and lacks intent comment for its safety-critical allowlist

Impact
The name reads as "any error from executing against downstream", but the function is a curated allowlist of exactly three terror codes that gate whether a checkpoint may advance past possibly non-durable DML.
A maintainer trusting the name could add or drop an error type, or assume other downstream DB failures are covered, and silently change checkpoint-durability behavior across all four call sites.

Scope

  • dm/syncer/execution_error.go:19 — isDownstreamExecutionError
  • dm/syncer/syncer.go:1294 — flushCheckPoints
  • dm/syncer/checkpoint_flush_worker.go:79 — checkpointFlushWorker.Run

Evidence
isDownstreamExecutionError matches only ErrDBExecuteFailed, ErrDBExecuteFailedBegin, and ErrDBUnExpect; any other downstream DB error returns false. The helper is now the single source of truth for the "skip checkpoint flush" decision but carries no doc comment explaining why these specific codes (and notably the newly added ErrDBExecuteFailedBegin) are the durability-relevant set.

Change request
Add a doc comment on isDownstreamExecutionError stating the invariant it encodes (these errors mean downstream DML may not be durable, so the checkpoint must not advance) and why ErrDBExecuteFailedBegin belongs here. Consider a more intent-revealing name (for example one that signals it is a checkpoint-skip predicate rather than a generic "any downstream execution error" check).

return err != nil &&
(terror.ErrDBExecuteFailed.Equal(err) ||
terror.ErrDBExecuteFailedBegin.Equal(err) ||
terror.ErrDBUnExpect.Equal(err))
}
6 changes: 3 additions & 3 deletions dm/syncer/syncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1291,7 +1291,7 @@ func (s *Syncer) flushCheckPoints() error {
// optimistic shard info, DM-master may resolved the optimistic lock and let other worker execute DDL. So after this
// worker resume, it can not execute the DML/DDL in old binlog because of downstream table structure mismatching.
// We should find a way to (compensating) implement a transaction containing interaction with both etcd and SQL.
if err != nil && (terror.ErrDBExecuteFailed.Equal(err) || terror.ErrDBUnExpect.Equal(err)) {
if isDownstreamExecutionError(err) {
s.tctx.L().Warn("error detected when executing SQL job, skip sync flush checkpoints",
zap.Stringer("checkpoint", s.checkpoint),
zap.Error(err))
Expand Down Expand Up @@ -1327,7 +1327,7 @@ func (s *Syncer) flushCheckPointsAsync(asyncFlushJob *job) {
// optimistic shard info, DM-master may resolved the optimistic lock and let other worker execute DDL. So after this
// worker resume, it can not execute the DML/DDL in old binlog because of downstream table structure mismatching.
// We should find a way to (compensating) implement a transaction containing interaction with both etcd and SQL.
if err != nil && (terror.ErrDBExecuteFailed.Equal(err) || terror.ErrDBUnExpect.Equal(err)) {
if isDownstreamExecutionError(err) {
s.tctx.L().Warn("error detected when executing SQL job, skip async flush checkpoints",
zap.Stringer("checkpoint", s.checkpoint),
zap.Error(err))
Expand Down Expand Up @@ -1974,7 +1974,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
}

// if any execute error, flush safemode exit point
if err2 = s.execError.Load(); err2 != nil && (terror.ErrDBExecuteFailed.Equal(err2) || terror.ErrDBUnExpect.Equal(err2)) {
if err2 = s.execError.Load(); isDownstreamExecutionError(err2) {
if err2 = s.checkpoint.FlushSafeModeExitPoint(s.tctx); err2 != nil {
s.tctx.L().Warn("failed to flush safe mode checkpoints when exit task", zap.Error(err2))
}
Expand Down
Loading