[DM] checkpoint may be flushed after downstream BeginTx failure and skip unapplied DML on resume

[GMHDBJD]

What did you do?

While investigating a DM incremental replication consistency issue, I found that checkpoint flushing does not treat downstream BeginTx failures as an execution error that should block checkpoint advancement.

A minimal local reproduction can be added on top of pingcap/tiflow upstream/master commit 43647d57a1d03070cd57344c4be9e4900d9e6cea:

go test ./dm/syncer -run TestCheckpointFlushWorkerSkipsCheckpointOnBeginError -count=1

The test sets the syncer's checkpoint flush worker execError to a downstream begin failure:

execError.Store(terror.ErrDBExecuteFailedBegin.Delegate(sql.ErrConnDone))

and then triggers a checkpoint flush. The test expects checkpoint flush to be skipped because the downstream DML transaction did not successfully begin, so pending DMLs may not be durable downstream.

Relevant code paths:

• dm/pkg/conn/baseconn.go: downstream BeginTx errors are wrapped as terror.ErrDBExecuteFailedBegin.
• dm/syncer/checkpoint_flush_worker.go: checkpoint flush is skipped only when execError matches terror.ErrDBExecuteFailed or terror.ErrDBUnExpect.
• dm/syncer/syncer.go: sync/async checkpoint flush uses the same skip predicate.

What did you expect to see?

Checkpoint flush should be skipped when a downstream SQL execution path fails in a way that means DMLs may not be durable downstream, including BeginTx failures such as sql.ErrConnDone / sql: connection is already closed.

After restart/resume, DM should replay from a checkpoint that is not beyond the unapplied DMLs.

What did you see instead?

The checkpoint flush worker still calls FlushPointsExcept after execError is set to terror.ErrDBExecuteFailedBegin.Delegate(sql.ErrConnDone).

The local regression test fails as follows:

--- FAIL: TestCheckpointFlushWorkerSkipsCheckpointOnBeginError (0.00s)
    checkpoint_flush_worker_repro_test.go:129:
        Error:          Not equal:
                        expected: 0
                        actual  : 1
        Messages:       checkpoint flush must be skipped after downstream BeginTx failure; flushing here can persist a checkpoint past non-durable DML
FAIL
FAIL github.com/pingcap/tiflow/dm/syncer 0.085s

This means a checkpoint can be persisted past DML jobs that failed before the downstream transaction was even created. On resume, DM can then start from the advanced checkpoint and skip those DMLs.

Impact

This is a data correctness risk for DM incremental replication. If the downstream connection is closed during BeginTx, some DMLs may not be applied to the downstream, but the checkpoint can still advance. After resume, the unapplied DMLs may not be replayed.

There is also a related diagnostic issue: judgeKeyNotFound can run after ExecuteSQL returns an error, which may produce misleading no matching record warnings for batches that did not actually execute successfully.

Suggested fix

• Treat terror.ErrDBExecuteFailedBegin as a checkpoint-blocking execution error in all checkpoint flush guards, together with terror.ErrDBExecuteFailed and terror.ErrDBUnExpect.
• Consider treating sql.ErrConnDone / sql: connection is already closed during BeginTx as retryable where safe.
• Avoid running key-not-found diagnostics after ExecuteSQL has already returned an execution error unless the execution result is valid.
• Add a regression test covering ErrDBExecuteFailedBegin(sql.ErrConnDone) and checkpoint flush behavior.

Versions of the cluster

DM version:

pingcap/tiflow upstream/master @ 43647d57a1d03070cd57344c4be9e4900d9e6cea

Upstream MySQL/MariaDB server version:

N/A for the minimal code-level reproduction

Downstream TiDB cluster version:

N/A for the minimal code-level reproduction

How did you deploy DM:

N/A for the minimal code-level reproduction

Other interesting information:

Observed when downstream transaction begin returned sql: connection is already closed.

Current status of DM cluster

N/A for the minimal code-level reproduction

[GMHDBJD]

Adding the local minimal reproduction diff I used to verify the issue on upstream/master.

This is only a reproducer/regression-test draft, not a complete fix. With current upstream/master, it fails because FlushPointsExcept is still called when execError is ErrDBExecuteFailedBegin(sql.ErrConnDone).

Run:

go test ./dm/syncer -run TestCheckpointFlushWorkerSkipsCheckpointOnBeginError -count=1

Diff:

diff --git a/dm/syncer/checkpoint_flush_worker_repro_test.go b/dm/syncer/checkpoint_flush_worker_repro_test.go
new file mode 100644
index 000000000..d1cce4730
--- /dev/null
+++ b/dm/syncer/checkpoint_flush_worker_repro_test.go
@@ -0,0 +1,133 @@
+// 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/go-mysql-org/go-mysql/mysql"
+	"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"
+)
+
+type reproCheckpoint struct {
+	flushCalls atomic.Int32
+}
+
+func (c *reproCheckpoint) Init(_ *tcontext.Context) error { return nil }
+func (c *reproCheckpoint) Close()                         {}
+func (c *reproCheckpoint) ResetConn(_ *tcontext.Context) error {
+	return nil
+}
+func (c *reproCheckpoint) Clear(_ *tcontext.Context) error { return nil }
+func (c *reproCheckpoint) Load(_ *tcontext.Context) error  { return nil }
+func (c *reproCheckpoint) LoadMeta(_ context.Context) error {
+	return nil
+}
+func (c *reproCheckpoint) SaveTablePoint(_ *filter.Table, _ binlog.Location, _ *model.TableInfo) {
+}
+func (c *reproCheckpoint) DeleteTablePoint(_ *tcontext.Context, _ *filter.Table) error {
+	return nil
+}
+func (c *reproCheckpoint) DeleteAllTablePoint(_ *tcontext.Context) error { return nil }
+func (c *reproCheckpoint) DeleteSchemaPoint(_ *tcontext.Context, _ string) error {
+	return nil
+}
+func (c *reproCheckpoint) IsOlderThanTablePoint(_ *filter.Table, _ binlog.Location) bool {
+	return false
+}
+func (c *reproCheckpoint) SaveGlobalPoint(_ binlog.Location)         {}
+func (c *reproCheckpoint) SaveGlobalPointForcibly(_ binlog.Location) {}
+func (c *reproCheckpoint) Snapshot(_ bool) *SnapshotInfo             { return nil }
+func (c *reproCheckpoint) DiscardPendingSnapshots()                  {}
+func (c *reproCheckpoint) FlushPointsExcept(_ *tcontext.Context, _ int, _ []*filter.Table, _ []string, _ [][]interface{}) error {
+	c.flushCalls.Inc()
+	return nil
+}
+func (c *reproCheckpoint) FlushPointsWithTableInfos(_ *tcontext.Context, _ []*filter.Table, _ []*model.TableInfo) error {
+	return nil
+}
+func (c *reproCheckpoint) FlushSafeModeExitPoint(_ *tcontext.Context) error { return nil }
+func (c *reproCheckpoint) GlobalPoint() binlog.Location {
+	return binlog.Location{Position: mysql.Position{Name: "mysql-bin.000001", Pos: 4}}
+}
+func (c *reproCheckpoint) GlobalPointSaveTime() time.Time           { return time.Time{} }
+func (c *reproCheckpoint) SaveSafeModeExitPoint(_ *binlog.Location) {}
+func (c *reproCheckpoint) SafeModeExitPoint() *binlog.Location      { return nil }
+func (c *reproCheckpoint) TablePoint() map[string]map[string]binlog.Location {
+	return nil
+}
+func (c *reproCheckpoint) GetTableInfo(_, _ string) *model.TableInfo { return nil }
+func (c *reproCheckpoint) FlushedGlobalPoint() binlog.Location {
+	return binlog.Location{Position: mysql.Position{Name: "mysql-bin.000001", Pos: 4}}
+}
+func (c *reproCheckpoint) LastFlushOutdated() bool { return false }
+func (c *reproCheckpoint) Rollback()               {}
+func (c *reproCheckpoint) String() string          { return "repro checkpoint" }
+func (c *reproCheckpoint) LoadIntoSchemaTracker(_ context.Context, _ *schema.Tracker) error {
+	return nil
+}
+func (c *reproCheckpoint) CheckAndUpdate(_ context.Context, _ map[string]string, _ map[string]map[string]string) error {
+	return nil
+}
+
+func TestCheckpointFlushWorkerSkipsCheckpointOnBeginError(t *testing.T) {
+	cp := &reproCheckpoint{}
+	execError := atomic.NewError(nil)
+	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)
+	}()
+
+	flushErrCh := make(chan error, 1)
+	worker.Add(&checkpointFlushTask{
+		snapshotInfo: &SnapshotInfo{
+			id:        1,
+			globalPos: binlog.Location{Position: mysql.Position{Name: "mysql-bin.000002", Pos: 4}},
+		},
+		syncFlushErrCh: flushErrCh,
+	})
+
+	require.NoError(t, <-flushErrCh)
+	worker.Close()
+	<-done
+
+	require.Equal(t, int32(0), cp.flushCalls.Load(),
+		"checkpoint flush must be skipped after downstream BeginTx failure; flushing here can persist a checkpoint past non-durable DML")
+}
+
+var _ CheckPoint = (*reproCheckpoint)(nil)