From f2c8a6faf4f1345c2863ac82c25fe5bf75766db5 Mon Sep 17 00:00:00 2001 From: gmhdbjd Date: Mon, 12 Oct 2026 12:27:10 +0000 Subject: [PATCH 1/2] worker(dm): isolate source errors in query status snapshots Deep-copy the saved source status while holding the server lock so repeated and concurrent queries cannot append errors to the shared result. Cover nil and existing saved errors, stable response size, and unchanged earlier snapshots. --- dm/worker/server.go | 5 ++- dm/worker/server_test.go | 74 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/dm/worker/server.go b/dm/worker/server.go index 5d3f813fa6..259a69c613 100644 --- a/dm/worker/server.go +++ b/dm/worker/server.go @@ -21,6 +21,7 @@ import ( "sync" "time" + "github.com/golang/protobuf/proto" "github.com/pingcap/errors" "github.com/pingcap/tidb/pkg/util" "github.com/pingcap/tiflow/dm/common" @@ -548,7 +549,9 @@ func (s *Server) getSourceStatus(needLock bool) pb.SourceStatus { s.mu.Lock() defer s.mu.Unlock() } - return s.sourceStatus + // QueryStatus appends the current source error to this snapshot. Do not share + // the saved Result, or repeated queries will accumulate errors in it. + return *proto.Clone(&s.sourceStatus).(*pb.SourceStatus) } // TODO: move some call to setWorker/getOrStartWorker. diff --git a/dm/worker/server_test.go b/dm/worker/server_test.go index 8740df6c79..b84a32dd06 100644 --- a/dm/worker/server_test.go +++ b/dm/worker/server_test.go @@ -725,6 +725,80 @@ func checkRelayStatus(cli pb.WorkerClient, expect pb.Stage) bool { return status.SourceStatus.RelayStatus.Stage == expect } +func TestQueryStatusDoesNotAccumulateSourceErrors(t *testing.T) { + for _, tc := range []struct { + name string + sourceErr error + errorCount int + }{ + {name: "no stored error", errorCount: 1}, + {name: "existing stored error", sourceErr: errors.New("failed to enable subtasks"), errorCount: 2}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := config.NewSourceConfig() + cfg.SourceID = "source-1" + w, err := NewSourceWorker(cfg, nil, "worker-1", "") + require.NoError(t, err) + w.closed.Store(false) + defer w.Stop(true) + + s := &Server{cfg: NewConfig(), worker: w} + s.setSourceStatus(cfg.SourceID, tc.sourceErr, true) + storedStatus := s.getSourceStatus(true) + storedBytes, err := storedStatus.Marshal() + require.NoError(t, err) + + // A missing upstream connection produces a source error on every query. + req := &pb.QueryStatusRequest{Name: "missing-task"} + var firstResponse *pb.QueryStatusResponse + var firstBytes []byte + checkResponse := func(resp *pb.QueryStatusResponse, err error) { + require.NoError(t, err) + require.True(t, resp.Result) + require.Equal(t, cfg.SourceID, resp.SourceStatus.Source) + require.Len(t, resp.SourceStatus.Result.Errors, tc.errorCount) + require.Equal(t, w.getSourceStatusErr(), resp.SourceStatus.Result.Errors[tc.errorCount-1]) + data, err := resp.Marshal() + require.NoError(t, err) + if firstResponse == nil { + firstResponse, firstBytes = resp, data + } else { + require.Equal(t, firstBytes, data) + } + } + for i := 0; i < 3; i++ { + checkResponse(s.QueryStatus(context.Background(), req)) + } + + // Concurrent queries must also own their error lists. + type queryResult struct { + response *pb.QueryStatusResponse + err error + } + results := make(chan queryResult, 8) + for i := 0; i < cap(results); i++ { + go func() { + resp, err := s.QueryStatus(context.Background(), req) + results <- queryResult{response: resp, err: err} + }() + } + for i := 0; i < cap(results); i++ { + result := <-results + checkResponse(result.response, result.err) + } + + // Neither the saved source status nor an earlier response may change. + storedStatus = s.getSourceStatus(true) + data, err := storedStatus.Marshal() + require.NoError(t, err) + require.Equal(t, storedBytes, data) + data, err = firstResponse.Marshal() + require.NoError(t, err) + require.Equal(t, firstBytes, data) + }) + } +} + func loadSourceConfigWithoutPassword(c *check.C) *config.SourceConfig { sourceCfg, err := config.SourceCfgFromYamlAndVerify(config.SampleSourceConfig) c.Assert(err, check.IsNil) From 107346889ac281b386df2e7415be5f10bcab8253 Mon Sep 17 00:00:00 2001 From: gmhdbjd Date: Mon, 12 Oct 2026 12:35:57 +0000 Subject: [PATCH 2/2] worker(dm): copy only the result before appending source errors Keep the source status getter unchanged. Clone the saved ProcessResult only when QueryStatus is about to append the current upstream error, following the existing result-copying pattern in getAllSubTaskStatus. --- dm/worker/server.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/dm/worker/server.go b/dm/worker/server.go index 259a69c613..3bbcca88c8 100644 --- a/dm/worker/server.go +++ b/dm/worker/server.go @@ -549,9 +549,7 @@ func (s *Server) getSourceStatus(needLock bool) pb.SourceStatus { s.mu.Lock() defer s.mu.Unlock() } - // QueryStatus appends the current source error to this snapshot. Do not share - // the saved Result, or repeated queries will accumulate errors in it. - return *proto.Clone(&s.sourceStatus).(*pb.SourceStatus) + return s.sourceStatus } // TODO: move some call to setWorker/getOrStartWorker. @@ -825,6 +823,9 @@ func (s *Server) QueryStatus(ctx context.Context, req *pb.QueryStatusRequest) (* if sourceErr := w.getSourceStatusErr(); sourceErr != nil { if resp.SourceStatus.Result == nil { resp.SourceStatus.Result = &pb.ProcessResult{} + } else { + // Do not append the current error to the saved source result. + resp.SourceStatus.Result = proto.Clone(resp.SourceStatus.Result).(*pb.ProcessResult) } resp.SourceStatus.Result.Errors = append(resp.SourceStatus.Result.Errors, sourceErr) }