dispatcher: duplicate WRITE for the same block event can be misreported as DONE while writer is still WRITING

[lidezhu]

What did you do?

When the writer dispatcher receives the first Action_Write, it accepts the action only in WAITING state, then immediately switches its local block stage to WRITING and starts the real downstream write asynchronously.

ticdc/downstreamadapter/dispatcher/helper.go

Lines 204 to 217 in ac1039f

	// actionMatchs checks whether the action is for the current pending ddl/sync point event.
	func (b *BlockEventStatus) actionMatchs(action *heartbeatpb.DispatcherAction) bool {
	b.mutex.Lock()
	defer b.mutex.Unlock()
	if b.blockPendingEvent == nil {
	return false
	}
	if b.blockStage != heartbeatpb.BlockStage_WAITING {
	return false
	}
	return b.blockCommitTs == action.CommitTs

ticdc/downstreamadapter/dispatcher/basic_dispatcher.go

Lines 826 to 849 in ac1039f

	if d.blockEventStatus.actionMatchs(action) {
	log.Info("pending event get the action",
	zap.Stringer("dispatcher", d.id),
	zap.Int64("mode", d.mode),
	zap.Any("action", action),
	zap.Any("innerAction", int(action.Action)),
	zap.Uint64("pendingEventCommitTs", pendingEvent.GetCommitTs()))
	actionCommitTs := action.CommitTs
	actionIsSyncPoint := action.IsSyncPoint
	d.blockEventStatus.updateBlockStage(heartbeatpb.BlockStage_WRITING)
	switch action.Action {
	case heartbeatpb.Action_Write:
	pendingEvent.PushFrontFlushFunc(func() {
	// clear blockEventStatus should be before wake ds.
	// otherwise, there may happen:
	// 1. wake ds
	// 2. get new ds and set new pending event
	// 3. clear blockEventStatus(should be the old pending event, but clear the new one)
	d.blockEventStatus.clear()
	})
	d.sharedInfo.GetBlockEventExecutor().Submit(d, func() {
	d.ExecuteBlockEventDDL(pendingEvent, actionCommitTs, actionIsSyncPoint)
	})
	return true

If the maintainer resends the same Action_Write before that asynchronous write has actually finished, the duplicate action no longer matches, because the dispatcher is already in WRITING instead of WAITING.
However, the dispatcher does not treat this as a harmless duplicate write. Instead, it falls through to the generic fallback path and still reports DONE for the same block event.

ticdc/downstreamadapter/dispatcher/basic_dispatcher.go

Lines 872 to 887 in ac1039f

	} else {
	ts, ok := d.blockEventStatus.getEventCommitTs()
	if ok && action.CommitTs > ts {
	log.Debug("pending event's commitTs is smaller than the action's commitTs, just ignore it",
	zap.Uint64("pendingEventCommitTs", ts),
	zap.Uint64("actionCommitTs", action.CommitTs),
	zap.Stringer("dispatcher", d.id))
	return false
	}
	}
	// Step4: whether the outdate message or not, we need to return message show we have finished the event.
	d.offerDoneBlockStatus(action.CommitTs, action.IsSyncPoint)
	}
	return false
	}

As a result, the maintainer can observe BlockStage_DONE even though the writer has not actually finished executing the DDL or syncpoint downstream. On the maintainer side, DONE is treated as a real completion signal and can immediately advance the barrier state machine.

• ticdc/maintainer/barrier.go

  Lines 364 to 365 in ac1039f

  	if status.State.Stage == heartbeatpb.BlockStage_DONE {
  	return b.handleEventDone(cfID, dispatcherID, status), nil, "", true

• ticdc/maintainer/barrier.go

  Lines 388 to 418 in ac1039f

  	if event.writerDispatcher == dispatcherID {
  	// Do not consume writer DONE until this route transition is ready. A false
  	// precheck may mean an earlier route event is still pending; real route
  	// conflicts are reported through routeAdmin.
  	if !b.precheckRouteEvent(event) {
  	return event
  	}
  	if !b.applyRouteEvent(event) {
  	return event
  	}
  	if event.needSchedule {
  	// we need do schedule when writerDispatcherAdvanced
  	// Otherwise, if we do schedule when just selected = true, then ask dispatcher execute ddl
  	// when meeting truncate table,
  	// there is possible that dml for the new table will arrive before truncate ddl executed.
  	// that will lead to data loss
  	scheduled := b.tryScheduleEvent(event)
  	if !scheduled {
  	// not scheduled yet, just return, wait for next resend
  	return event
  	}
  	} else {
  	// the pass action will be sent periodically in resend logic if not acked
  	event.writerDispatcherAdvanced = true
  	event.lastResendTime = time.Now().Add(-20 * time.Second)
  	}
  	}
  	// checkpoint ts is advanced, clear the map, so do not need to resend message anymore
  	event.markDispatcherEventDone(dispatcherID)
  	b.checkEventFinish(event)

In short, the bug is that the current implementation conflates two different states:

• the writer has already accepted the WRITE action
• the writer has actually finished the write

Under duplicate WRITE resend timing, the first state can be incorrectly reported as the second one.

The unconditional fallback DONE path existed earlier, but it was not practically harmful while Action_Write was executed synchronously in HandleDispatcherStatus, because the dispatcher could not process a duplicate WRITE for the same event until the first write had already completed.

The bug became reachable after PR #3646 made block-event writes asynchronous. From that point on, the dispatcher could enter local WRITING state, return to the status handling path, and then process a duplicate WRITE for the same (commitTs, isSyncPoint) before the original write had actually finished. Because unmatched actions still fell through to the generic fallback DONE path, the duplicate WRITE could now be misreported as real completion.

What did you expect to see?

No response

What did you see instead?

As described above

Versions of the cluster

Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

(paste TiDB cluster version here)

Upstream TiKV version (execute tikv-server --version):

(paste TiKV version here)

TiCDC version (execute cdc version):

(paste TiCDC version here)

[wk989898]

According to the description, the behavior here is to treat all events smaller than the pending event as completed, while in other cases, it returns false. Is it possible to receive an event smaller than the pending event here?

However, the dispatcher does not treat this as a harmless duplicate write. Instead, it falls through to the generic fallback path and still reports DONE for the same block event.

ticdc/downstreamadapter/dispatcher/basic_dispatcher.go

Lines 872 to 887 in ac1039f

	} else {
	ts, ok := d.blockEventStatus.getEventCommitTs()
	if ok && action.CommitTs > ts {
	log.Debug("pending event's commitTs is smaller than the action's commitTs, just ignore it",
	zap.Uint64("pendingEventCommitTs", ts),
	zap.Uint64("actionCommitTs", action.CommitTs),
	zap.Stringer("dispatcher", d.id))
	return false
	}
	}
	// Step4: whether the outdate message or not, we need to return message show we have finished the event.
	d.offerDoneBlockStatus(action.CommitTs, action.IsSyncPoint)
	}
	return false
	}