syncer(dm): bound graceful stop during async DDL polling - #12785
Conversation
Arm the graceful-stop deadline before waiting for the transaction lock and make asynchronous DDL status polling follow the graceful run context. This lets Pause and Stop interrupt the polling cycle without canceling ordinary DDL execution. Retain enough in-memory DDL identity to reconcile the downstream job on same-instance Resume before advancing the checkpoint, and add lifecycle-level race coverage for cancellation and recovery.
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
do-not-merge/work-in-progress
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
ti-chi-bot
Bot
added
area/dm
ti-chi-bot
Bot
added
do-not-merge/needs-triage-completed
release-note
📝 WalkthroughWalkthroughThe syncer now tracks unresolved asynchronous DDLs across pause and resume, polls them with context cancellation, reconciles their outcomes before later DDL execution, and coordinates graceful shutdown with active polling and transaction locks. ChangesAsync DDL lifecycle
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related issues
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (6)
dm/syncer/syncer_test.go (2)
2100-2103: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe wall-clock lower bound can flake on a loaded runner.
stopDurationmust be at leastwaitOnStop/2, which is 100ms against a 200ms budget.waitBeforeRunExitsubtractsprepareForWaitTimefromwaitDuration, so a slowrefreshCliArgsshortens the real wait. The margin is 50%, which is usually enough, but lower-bound timing assertions are a common source of CI flakes.Consider enabling the existing
recordAndIgnorePrepareTimefailpoint, asTestWaitBeforeRunExitdoes, so the prepare time is excluded and the bound becomes deterministic.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/syncer_test.go` around lines 2100 - 2103, Update the test around the stopDuration assertions to enable the existing recordAndIgnorePrepareTime failpoint, matching TestWaitBeforeRunExit, before invoking the wait flow. Ensure prepare time is excluded from waitDuration so the lower-bound assertion remains deterministic while preserving the existing upper-bound check.
2113-2117: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe embedded nil
CheckPointmakes future changes fail loudly and unhelpfully.
asyncDDLBarrierCheckpointembeds theCheckPointinterface without a value. The four overridden methods happen to cover every call the DDL path makes today. If a later change adds another checkpoint call on that path, the call dispatches to a nil interface and panics inside thesyncDDLgoroutine, which aborts the whole test binary instead of failing this test.Consider embedding a real
NewRemoteCheckPoint(...)value, or adding a short comment that lists the methods the DDL path relies on.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/syncer_test.go` around lines 2113 - 2117, Update asyncDDLBarrierCheckpoint so its embedded CheckPoint is initialized with a real NewRemoteCheckPoint(...) value instead of a nil interface, while preserving the existing method overrides and synchronization behavior.dm/syncer/syncer.go (2)
1527-1531: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueIgnored DDL jobs now get the timestamp and timezone wrappers.
When
ignoreis true and a pending reconcile descriptor exists, the block still appendsSET TIMESTAMPandSET SESSION TIME_ZONEtoddlJob.ddlsand updatess.timezoneLastTime. The wrapped list is only needed for thematchescomparison, but it also changes theddlslogged by the shard-DDL handlers at lines 1634 and 1647. This is cosmetic, and the barrier behavior itself is correct.If you prefer to keep ignored jobs unmodified, build the wrapped list in a local variable and pass that to
reconcileAsyncDDL.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/syncer.go` around lines 1527 - 1531, The conditional in the async DDL reconciliation flow correctly preserves the barrier, but mutates ignored jobs’ `ddlJob.ddls` with timestamp and timezone wrappers. Keep `ddlJob.ddls` unchanged when `ignore` is true by constructing the wrapped statements in a local variable for `matches` and passing that variable to `reconcileAsyncDDL`, while preserving normal behavior for non-ignored jobs.
1752-1797: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe graceful-deadline rework is correct. Two consistency points.
The timer ordering, the
timeoutDonehandshake, and the deferredtimer.Stop()/<-timeoutDonedrain are sound.timer.Stop()is called exactly once, so afalseresult always means the callback fired and will closetimeoutDone. The defer cannot block indefinitely.
- Line 1791 repeats the nil check that
forceStopalready performs. CallforceStop()for one cancellation path.- Line 1724 captures
runCancelbecauses.runCancelcan be replaced by a later Resume. Line 1810 still reads the mutables.runCtxfield. The goroutine is joined throughrunWgbeforeRunreturns, so this is not reachable today, but capturing both values keeps the stated invariant complete.♻️ Proposed changes
runCancel := s.runCancel + runCtx := s.runCtx forceStop := func() { if runCancel != nil { runCancel() } }if s.isTransactionEnd { s.waitXIDJob.Store(int64(waitComplete)) s.waitTransactionLock.Unlock() s.tctx.L().Info("the last job is transaction end, done directly") - if runCancel != nil { - runCancel() - } + forceStop() return }select { - case <-s.runCtx.Ctx.Done(): + case <-runCtx.Ctx.Done(): s.tctx.L().Info("syncer run exit so runCtx done") case <-timeoutDone: }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/syncer.go` around lines 1752 - 1797, Update the timeout completion path around forceStop and the transaction-end branch: remove the redundant runCancel nil check and invoke forceStop directly, then capture the current s.runCtx alongside runCancel before launching the goroutine and use that captured context at the later s.runCtx reference, preserving the invariant that Resume cannot replace values used by the running operation.dm/syncer/error.go (2)
126-137: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider making the clear sequence atomic.
clearResolvedAsyncDDLtakesasyncDDLReconcileMuthree times: ingetAsyncDDLReconcileInfo, inasyncDDLResolved, and inclearAsyncDDLReconcileInfo. Another goroutine can replaces.asyncDDLReconcilebetween those calls. The pointer identity check insideclearAsyncDDLReconcileInfokeeps the outcome safe today, so this is a robustness improvement rather than a live defect. A single locked helper removes the check-then-act window.♻️ Proposed refactor
func (s *Syncer) clearResolvedAsyncDDL( ddls []string, startLocation binlog.Location, currentLocation binlog.Location, ) { - info := s.getAsyncDDLReconcileInfo() - if info == nil || !s.asyncDDLResolved(info) || - !info.matches(ddls, startLocation, currentLocation, s.cfg.EnableGTID) { - return - } - s.clearAsyncDDLReconcileInfo(info) + s.asyncDDLReconcileMu.Lock() + defer s.asyncDDLReconcileMu.Unlock() + info := s.asyncDDLReconcile + if info == nil || !info.resolved || + !info.matches(ddls, startLocation, currentLocation, s.cfg.EnableGTID) { + return + } + s.asyncDDLReconcile = nil }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/error.go` around lines 126 - 137, Make clearResolvedAsyncDDL’s read/validate/clear sequence atomic by adding or reusing a helper that acquires asyncDDLReconcileMu once, checks the reconcile info and matches conditions while holding the lock, then clears the same state before unlocking. Update clearResolvedAsyncDDL to use this helper instead of separately calling getAsyncDDLReconcileInfo, asyncDDLResolved, and clearAsyncDDLReconcileInfo.
189-215: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winA transient status-query failure ends polling and reports the original error.
If
getDDLStatusFromTiDBreturns an error,statuskeeps its zero value"". The switch then reachesdefaultand returnsoriginErr. One temporary downstream query failure therefore aborts reconciliation and surfacesinvalid connection, even though the TiDB job may still be running. The context is still live in that case, so the loop could instead wait for the next tick and retry.This preserves the previous inline behavior, so it is not a regression. Consider making the retry explicit while this block is being rewritten.
Also, line 194 and line 198 use
s.tctx.L()while line 213 usestctx.L(). Use one logger consistently in this function.♻️ Proposed change for the transient-error path
status, err := getDDLStatusFromTiDB(tctx, conn, ddl, createTime) if err != nil { if ctxErr := tctx.Ctx.Err(); ctxErr != nil { return ctxErr } - s.tctx.L().Warn("error when getting DDL status from TiDB", zap.Error(err)) + tctx.L().Warn("error when getting DDL status from TiDB, will retry", zap.Error(err)) + select { + case <-tctx.Ctx.Done(): + return tctx.Ctx.Err() + case <-ticker.C: + } + continue }Note: the
TestStatusfailpoint injects a status after the query, so keep that injection reachable if any test relies on it together with a failing query.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dm/syncer/error.go` around lines 189 - 215, Update the polling block around getDDLStatusFromTiDB so a live-context query error skips status evaluation and retries on the next tick instead of falling through to the default case and returning originErr; preserve the existing context-cancellation return and keep the TestStatus failpoint reachable. Use one consistent logger, matching the logger used by the surrounding tctx flow, for all messages in this block.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@dm/syncer/error_test.go`:
- Around line 202-206: Fix the assertion after handleSpecialDDLError in the
relevant test so it verifies the returned handledErr rather than treating it as
a message. Assert that handledErr matches the expected execErr using the
appropriate error-comparison assertion, while retaining the existing mock
expectation check.
In `@dm/syncer/error.go`:
- Around line 226-250: Update dm/syncer/error.go lines 226-250: rename the
reconcileAsyncDDL boolean result to skipExecution and revise its documentation
to state it is true when a pending DDL matches or exists but does not match.
Update dm/syncer/syncer.go lines 1559-1563 to guard db.ExecuteSQLWithIgnore with
!ignore, !reconciled, and reconcileErr == nil, making execution impossible after
reconciliation errors.
- Around line 430-444: Preserve the timezone associated with pending async DDL
across Pause/Resume: when creating or recording the barrier in the async DDL
reconciliation flow, store the effective timezone alongside the existing
reconcile information, and restore it during Resume/reset before retrying the
DDL. Update the relevant async DDL state helpers such as
setAsyncDDLReconcileInfo and clear/restore handling so DDL jobs without
ddlJob.timezone reuse the original timezone.
---
Nitpick comments:
In `@dm/syncer/error.go`:
- Around line 126-137: Make clearResolvedAsyncDDL’s read/validate/clear sequence
atomic by adding or reusing a helper that acquires asyncDDLReconcileMu once,
checks the reconcile info and matches conditions while holding the lock, then
clears the same state before unlocking. Update clearResolvedAsyncDDL to use this
helper instead of separately calling getAsyncDDLReconcileInfo, asyncDDLResolved,
and clearAsyncDDLReconcileInfo.
- Around line 189-215: Update the polling block around getDDLStatusFromTiDB so a
live-context query error skips status evaluation and retries on the next tick
instead of falling through to the default case and returning originErr; preserve
the existing context-cancellation return and keep the TestStatus failpoint
reachable. Use one consistent logger, matching the logger used by the
surrounding tctx flow, for all messages in this block.
In `@dm/syncer/syncer_test.go`:
- Around line 2100-2103: Update the test around the stopDuration assertions to
enable the existing recordAndIgnorePrepareTime failpoint, matching
TestWaitBeforeRunExit, before invoking the wait flow. Ensure prepare time is
excluded from waitDuration so the lower-bound assertion remains deterministic
while preserving the existing upper-bound check.
- Around line 2113-2117: Update asyncDDLBarrierCheckpoint so its embedded
CheckPoint is initialized with a real NewRemoteCheckPoint(...) value instead of
a nil interface, while preserving the existing method overrides and
synchronization behavior.
In `@dm/syncer/syncer.go`:
- Around line 1527-1531: The conditional in the async DDL reconciliation flow
correctly preserves the barrier, but mutates ignored jobs’ `ddlJob.ddls` with
timestamp and timezone wrappers. Keep `ddlJob.ddls` unchanged when `ignore` is
true by constructing the wrapped statements in a local variable for `matches`
and passing that variable to `reconcileAsyncDDL`, while preserving normal
behavior for non-ignored jobs.
- Around line 1752-1797: Update the timeout completion path around forceStop and
the transaction-end branch: remove the redundant runCancel nil check and invoke
forceStop directly, then capture the current s.runCtx alongside runCancel before
launching the goroutine and use that captured context at the later s.runCtx
reference, preserving the invariant that Resume cannot replace values used by
the running operation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ba029f71-a606-433f-bffc-95b11186142b
📒 Files selected for processing (4)
dm/syncer/error.godm/syncer/error_test.godm/syncer/syncer.godm/syncer/syncer_test.go
| handledErr = syncer.handleSpecialDDLError( | ||
| tctx, execErr, ddls, 0, conn2, -1, binlog.Location{}, binlog.Location{}) | ||
| require.NoError(t, mock.ExpectationsWereMet()) | ||
| require.Error(t, execErr, handledErr) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Line 205 asserts nothing about handledErr.
require.Error(t, execErr, handledErr) checks that execErr is non-nil and uses handledErr as the message argument. execErr is constructed non-nil at line 165, so the assertion always passes and the returned handledErr is never verified. Compare the two errors instead.
💚 Proposed fix
- require.NoError(t, mock.ExpectationsWereMet())
- require.Error(t, execErr, handledErr)
+ require.NoError(t, mock.ExpectationsWereMet())
+ require.Equal(t, execErr, handledErr)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| handledErr = syncer.handleSpecialDDLError( | |
| tctx, execErr, ddls, 0, conn2, -1, binlog.Location{}, binlog.Location{}) | |
| require.NoError(t, mock.ExpectationsWereMet()) | |
| require.Error(t, execErr, handledErr) | |
| } | |
| handledErr = syncer.handleSpecialDDLError( | |
| tctx, execErr, ddls, 0, conn2, -1, binlog.Location{}, binlog.Location{}) | |
| require.NoError(t, mock.ExpectationsWereMet()) | |
| require.Equal(t, execErr, handledErr) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dm/syncer/error_test.go` around lines 202 - 206, Fix the assertion after
handleSpecialDDLError in the relevant test so it verifies the returned
handledErr rather than treating it as a message. Assert that handledErr matches
the expected execErr using the appropriate error-comparison assertion, while
retaining the existing mock expectation check.
| // reconcileAsyncDDL resolves an asynchronous DDL left by a previous paused | ||
| // run. The bool result reports whether a pending DDL matched the current job, | ||
| // in which case the caller must not submit the DDL again. | ||
| func (s *Syncer) reconcileAsyncDDL( | ||
| tctx *tcontext.Context, | ||
| ddls []string, | ||
| startLocation binlog.Location, | ||
| currentLocation binlog.Location, | ||
| conn *dbconn.DBConn, | ||
| ) (bool, error) { | ||
| info := s.getAsyncDDLReconcileInfo() | ||
| if info == nil { | ||
| return false, nil | ||
| } | ||
| if !info.matches(ddls, startLocation, currentLocation, s.cfg.EnableGTID) { | ||
| s.tctx.L().Error("pending asynchronous DDL does not match the current DDL", | ||
| zap.Strings("pending DDLs", info.ddls), | ||
| zap.Strings("current DDLs", ddls), | ||
| zap.Stringer("pending start location", info.startLocation), | ||
| zap.Stringer("current start location", startLocation), | ||
| zap.Stringer("pending end location", info.currentLocation), | ||
| zap.Stringer("current end location", currentLocation)) | ||
| // Do not execute a later DDL past an unresolved DDL barrier. | ||
| return true, terror.ErrDBUnExpect.Generate("pending asynchronous DDL does not match the current DDL") | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
The reconcileAsyncDDL bool has an implicit contract that both the definition and the caller depend on. The function returns true on the mismatch path even though its doc comment says the bool reports whether a pending DDL "matched". The caller then depends on the unstated invariant that the bool is always true when the error is non-nil, because that is the only thing that keeps db.ExecuteSQLWithIgnore from running after a failed reconciliation.
dm/syncer/error.go#L226-L250: rename the result toskipExecutionand update the doc comment to state that it is true both when a pending DDL matches and when a pending DDL exists but does not match.dm/syncer/syncer.go#L1559-L1563: make the dependency explicit, for example by guarding execution withif !ignore && !reconciled && reconcileErr == nil, so a future change to the reconciler cannot silently re-enable execution on an error path.
📍 Affects 2 files
dm/syncer/error.go#L226-L250(this comment)dm/syncer/syncer.go#L1559-L1563
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dm/syncer/error.go` around lines 226 - 250, Update dm/syncer/error.go lines
226-250: rename the reconcileAsyncDDL boolean result to skipExecution and revise
its documentation to state it is true when a pending DDL matches or exists but
does not match. Update dm/syncer/syncer.go lines 1559-1563 to guard
db.ExecuteSQLWithIgnore with !ignore, !reconciled, and reconcileErr == nil,
making execution impossible after reconciliation errors.
| if len(ddls) == 0 || index < 0 || index > len(ddls)-1 || errors.Cause(err) != mysql.ErrInvalidConn || createTime == -1 { | ||
| return err // return the original error | ||
| } | ||
|
|
||
| duration := 30 | ||
| failpoint.Inject("ChangeDuration", func() { | ||
| duration = 1 | ||
| }) | ||
| ticker := time.NewTicker(time.Duration(duration) * time.Second) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| status, err2 := getDDLStatusFromTiDB(tctx, conn, ddls[index], createTime) | ||
| if err2 != nil { | ||
| s.tctx.L().Warn("error when getting DDL status from TiDB", zap.Error(err2)) | ||
| } | ||
| failpoint.Inject("TestStatus", func(val failpoint.Value) { | ||
| status = val.(string) | ||
| s.tctx.L().Info("injected test status:", zap.String("TestStatus", status)) | ||
| }) | ||
| switch status { | ||
| case model.JobStateDone.String(), model.JobStateSynced.String(): | ||
| return nil | ||
| case model.JobStateCancelled.String(), model.JobStateRollingback.String(), model.JobStateRollbackDone.String(), model.JobStateCancelling.String(): | ||
| return terror.ErrSyncerCancelledDDL.Generate(ddls[index]) | ||
| case model.JobStateRunning.String(), model.JobStateQueueing.String(), model.JobStateNone.String(): | ||
| default: | ||
| tctx.L().Warn("Unexpected DDL status", zap.String("DDL status", status)) | ||
| return err | ||
| } | ||
| select { | ||
| case <-tctx.Ctx.Done(): | ||
| return err | ||
| case <-ticker.C: | ||
| } | ||
| info := s.setAsyncDDLReconcileInfo(ddls, index, createTime, startLocation, currentLocation) | ||
| retErr := s.pollAsyncDDL(s.asyncDDLPollContext(tctx), err, info.ddls[info.index], conn, info.createTime) | ||
| if retErr == nil { | ||
| s.markAsyncDDLResolved(info) | ||
| } else if terror.ErrSyncerCancelledDDL.Equal(retErr) { | ||
| // A terminal cancelled/rollback state is a known failure rather than | ||
| // an unknown async result. Surface it now and allow an explicit Resume | ||
| // to use the syncer's normal retry/recovery path. | ||
| s.clearAsyncDDLReconcileInfo(info) | ||
| } | ||
| return retErr |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect timezoneLastTime lifecycle and reset() behavior in the syncer.
rg -nP --type=go -C4 '\btimezoneLastTime\b'
ast-grep run --pattern $'func (s *Syncer) reset() {
$$$
}' --lang go dm/syncer/syncer.goRepository: pingcap/tiflow
Length of output: 152
🏁 Script executed:
#!/bin/bash
set -u
echo "== git status =="
git status --short
echo "== candidate error.go lines =="
fd -a 'error\.go$' . | sed 's#^\./##' | head -50
if [ -f dm/syncer/error.go ]; then
sed -n '400,460p' dm/syncer/error.go | nl -ba -v400
fi
echo "== timezoneLastTime occurrences =="
rg -nP --type=go -C3 '\btimezoneLastTime\b' . || true
echo "== reset occurrences =="
rg -n --type=go -C4 'func \(s \*Syncer\) reset\(|reset\(\)' dm/syncer || trueRepository: pingcap/tiflow
Length of output: 9990
🏁 Script executed:
#!/bin/bash
set -u
echo "== syncer.go reset section =="
sed -n '678,770p' dm/syncer/syncer.go
echo "== syncer.go asyncDDL reconcile related sections =="
rg -n --type=go -C6 'asyncDDLBarrierCheckpoint|ErrDBUnExpect|setAsyncDDLReconcileInfo|getAsyncDDLReconcileInfo|clearResolvedAsyncDDL|reconcileAsyncDDL|pollAsyncDDL|waitAsyncDDL' dm/syncer/syncer.go dm/syncer/error.go
echo "== error.go relevant async reconcilation =="
sed -n '400,470p' dm/syncer/error.go
sed -n '1,80p' dm/syncer/error.goRepository: pingcap/tiflow
Length of output: 22751
🏁 Script executed:
#!/bin/bash
set -u
echo "** dm/syncer/error.go reset and async info implementations **"
sed -n '1,280p' dm/syncer/error.go
echo "** dm/syncer/syncer.go tests around async DDL barrier and reset =="
sed -n '2180,2270p' dm/syncer/syncer_test.go
echo "** dm/syncer/syncer.go resume/restart path references =="
rg -n --type=go -C4 'func \(s \*Syncer\) Resume|Resume|resetDBs|Process\(' dm/syncer/syncer.goRepository: pingcap/tiflow
Length of output: 14946
🏁 Script executed:
#!/bin/bash
set -u
echo "** NewSyncer and timezoneLastTime init sites =="
rg -n --type=go -C5 'NewSyncer|--type=go|timezoneLastTime|time.LastTime|DefaultTimeZone|time.Last|s\.timezone|ddlJob\.timezone|ddlJob.timestamp' dm/syncer dm/dbconn dm/config dm/util 2>/dev/null || true
echo "** syncer.go struct and initialization around timezone field =="
sed -n '160,230p' dm/syncer/syncer.go
rg -n --type=go -C3 'func NewSyncer|type Syncer struct' dm/syncer/syncer.go dm/syncer -g '*.go' || trueRepository: pingcap/tiflow
Length of output: 50370
Preserve the async DDL barrier timezone after Resume.
reset() clears most in-memory syncer state but does not restore s.timezoneLastTime. When a DDL job without ddlJob.timezone is retried after Resume, it can use the reset default instead of the timezone stored when the barrier was created, so wrapped DDL SQL strings differ and reconcileAsyncDDL returns the permanent ErrDBUnExpect barrier. Store/restore the timezone used for the pending DDL during Pause/Resume.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dm/syncer/error.go` around lines 430 - 444, Preserve the timezone associated
with pending async DDL across Pause/Resume: when creating or recording the
barrier in the async DDL reconciliation flow, store the effective timezone
alongside the existing reconcile information, and restore it during Resume/reset
before retrying the DDL. Update the relevant async DDL state helpers such as
setAsyncDDLReconcileInfo and clear/restore handling so DDL jobs without
ddlJob.timezone reuse the original timezone.
|
@GMHDBJD: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
1 participant
What problem does this PR solve?
When a downstream TiDB DDL continues asynchronously after its SQL connection
returns
mysql.ErrInvalidConn, DM polls the TiDB DDL job fromwaitAsyncDDL.handleJobholdswaitTransactionLockwhile waiting for the DDL worker, butwaitBeforeRunExitpreviously needed the same lock before it could reach thegraceful-stop timeout path. The DDL poller also used
syncCtx, which wascanceled only after
runWg.Wait(). Together these dependencies allowedPause/Stop to wait for the downstream DDL for hours instead of honoring
wait-time-on-stop.Issue Number: close #12781
What is changed and how it works?
waitTransactionLock, so thedeadline can fire while
handleJobowns the lock.late callback from canceling a subsequent Resume.
runCtxfor asynchronous DDL status polling andreconciliation. Regular DDL execution remains on
syncCtx, so reaching thegraceful deadline does not cancel an in-progress
ExecuteSQLand create anew unknown DDL submission result.
original
mysql.ErrInvalidConnas a fatal downstream error.routed/wrapped SQL list, failed statement index, TiDB DDL creation time, and
binlog start/end locations.
checkpoint;
instead of executing another DDL past the unresolved job.
flushing succeed.
The descriptor is intentionally scoped to Pause/Stop and Resume on the same
Syncerwith unchanged DDL processing configuration. This change does notpersist the descriptor across worker restarts or add a global barrier for
configuration or
handle-errorchanges.Check List
Tests
make dm_unit_test_pkg PKG=github.com/pingcap/tiflow/dm/syncergo test -race ./dm/syncer -run 'Test(WaitAsyncDDLCanceled|ReconcileAsyncDDLMismatchIsBarrier|WaitBeforeRunExitCancelsBlockedAsyncDDL|SyncDDLReconcilesPendingAsyncDDL)$' -count=1go test -race ./dm/syncer -run '^TestWaitBeforeRunExitCancelsBlockedAsyncDDL$' -count=10go vet ./dm/syncermake fmtgit diff --checkQuestions
Will it cause performance regression or break compatibility?
No user-facing API or configuration compatibility change is expected. The
additional descriptor and status matching are used only after the existing
asynchronous-DDL invalid-connection recovery path and its subsequent Resume.
A terminal cancelled DDL remains an explicit task error and does not advance
the checkpoint.
Do you need to update user documentation, design documentation or monitoring documentation?
No.
Release note
Summary by CodeRabbit