Skip to content

eventbroker: fix broker quota release and message validation #5643

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
lidezhu wants to merge 2 commits into
base: master
Choose a base branch
from
Open

eventbroker: fix broker quota release and message validation #5643

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ac0e78f
eventservice: fix broker quota release and message validation
lidezhu Jul 12, 2026
47a25dc
fix
lidezhu Jul 12, 2026
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
14 changes: 8 additions & 6 deletions pkg/eventservice/event_broker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -706,6 +706,7 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) {
}

if uint64(sl.maxDMLBytes) > task.availableMemoryQuota.Load() {
releaseQuota(available, uint64(sl.maxDMLBytes))

@gemini-code-assist gemini-code-assist Bot Jul 12, 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.

medium

While releasing the allocated quota here is necessary, a similar quota leak can occur if scanner.scan fails further down in this function (around line 728). When scanner.scan returns an error, the remaining scannedBytes of the allocated quota is not fully released, which can lead to temporary quota exhaustion until the next congestion control update.

Consider fully releasing the remaining quota in the error handling block of scanner.scan as well:

if err != nil {
    if scannedBytes > 0 {
        releaseQuota(available, uint64(min(sl.maxDMLBytes, scannedBytes)))
    }
    log.Error("scan events failed", ...)
    return
}

log.Debug("dispatcher available memory quota is not enough, skip scan", zap.Stringer("dispatcher", task.id), zap.Uint64("available", task.availableMemoryQuota.Load()), zap.Int64("required", int64(sl.maxDMLBytes)))
c.sendSignalResolvedTs(task)
metrics.EventServiceSkipScanCount.WithLabelValues("dispatcher_quota").Inc()
Expand All @@ -714,24 +715,24 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) {

scanner := newEventScanner(c.eventStore, c.schemaStore, c.mounter, task.info.GetMode())
scannedBytes, events, interrupted, err := scanner.scan(ctx, task, dataRange, sl)
if scannedBytes < 0 {
releaseQuota(available, uint64(sl.maxDMLBytes))
} else if scannedBytes >= 0 && scannedBytes < sl.maxDMLBytes {
releaseQuota(available, uint64(sl.maxDMLBytes-scannedBytes))
}

if interrupted {
metrics.EventServiceInterruptScanCount.Inc()
}

if err != nil {
releaseQuota(available, uint64(sl.maxDMLBytes))
log.Error("scan events failed",
zap.Stringer("changefeedID", task.changefeedStat.changefeedID),
zap.Stringer("dispatcherID", task.id), zap.Int64("tableID", task.info.GetTableSpan().GetTableID()),
zap.Any("dataRange", dataRange), zap.Uint64("receivedResolvedTs", task.receivedResolvedTs.Load()),
zap.Uint64("sentResolvedTs", task.sentResolvedTs.Load()), zap.Error(err))
return
}
if scannedBytes < 0 {
releaseQuota(available, uint64(sl.maxDMLBytes))
} else if scannedBytes < sl.maxDMLBytes {
releaseQuota(available, uint64(sl.maxDMLBytes-scannedBytes))
}

if scannedBytes > int64(c.scanLimitInBytes) {
log.Info("scan bytes exceeded the limit, there must be a big transaction", zap.Stringer("dispatcher", task.id), zap.Int64("scannedBytes", scannedBytes), zap.Int64("limit", int64(c.scanLimitInBytes)))
Expand Down Expand Up @@ -1017,6 +1018,7 @@ func (c *eventBroker) addDispatcher(info DispatcherInfo) error {
status.addDispatcher(id, dispatcherPtr)
if span.Equal(common.KeyspaceDDLSpan(span.KeyspaceID)) {
c.tableTriggerDispatchers.Store(id, dispatcherPtr)
c.metricsCollector.metricDispatcherCount.Inc()
log.Info("table trigger dispatcher register dispatcher",
zap.Uint64("clusterID", c.tidbClusterID),
zap.Stringer("changefeedID", changefeedID),
Expand Down
71 changes: 71 additions & 0 deletions pkg/eventservice/event_broker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,11 @@ import (
"github.com/pingcap/ticdc/pkg/filter"
"github.com/pingcap/ticdc/pkg/integrity"
"github.com/pingcap/ticdc/pkg/messaging"
"github.com/pingcap/ticdc/pkg/metrics"
"github.com/pingcap/ticdc/pkg/node"
"github.com/pingcap/ticdc/pkg/pdutil"
"github.com/pingcap/ticdc/pkg/util"
"github.com/prometheus/client_golang/prometheus/testutil"
"github.com/stretchr/testify/require"
"github.com/tikv/client-go/v2/oracle"
"go.uber.org/atomic"
Expand Down Expand Up @@ -204,6 +206,75 @@ func TestAddDispatcherUnregisterOnSchemaStoreError(t *testing.T) {
require.Equal(t, uint64(1), es.unregisterCount.Load())
}

func TestDoScanReleasesChangefeedQuotaOnDispatcherQuotaFailure(t *testing.T) {
broker, _, _, _ := newEventBrokerForTest()
defer broker.close()

info := newMockDispatcherInfoForTest(t)
info.epoch = 1
status := broker.getOrSetChangefeedStatus(info)

disp := newDispatcherStat(info, 1, 1, nil, status)
disp.receivedResolvedTs.Store(102)
disp.eventStoreCommitTs.Store(101)
disp.availableMemoryQuota.Store(minScanLimitInBytes - 1)

serverID := node.ID(info.GetServerID())
changefeedQuota := atomic.NewUint64(minScanLimitInBytes * 2)
status.availableMemoryQuota.Store(serverID, changefeedQuota)

broker.doScan(context.Background(), disp)

require.Equal(t, uint64(minScanLimitInBytes*2), changefeedQuota.Load())
}

func TestDoScanReleasesChangefeedQuotaOnScanError(t *testing.T) {
broker, eventStore, schemaStore, _ := newEventBrokerForTest()
defer broker.close()

info := newMockDispatcherInfoForTest(t)
info.epoch = 1
require.NoError(t, broker.addDispatcher(info))

disp := broker.getDispatcher(info.GetID()).Load()
require.NotNil(t, disp)
disp.receivedResolvedTs.Store(102)
disp.eventStoreCommitTs.Store(101)
disp.availableMemoryQuota.Store(minScanLimitInBytes * 2)

status := broker.getOrSetChangefeedStatus(info)
serverID := node.ID(info.GetServerID())
changefeedQuota := atomic.NewUint64(minScanLimitInBytes * 2)
status.availableMemoryQuota.Store(serverID, changefeedQuota)

schemaStore.getTableInfoError = errors.New("mock get table info error")
require.NoError(t, eventStore.AppendEvents(info.GetID(), 102, &common.RawKVEntry{
StartTs: 101,
CRTs: 101,
Key: []byte("key"),
Value: []byte("value"),
}))

broker.doScan(context.Background(), disp)

require.Equal(t, uint64(minScanLimitInBytes*2), changefeedQuota.Load())
}

func TestTableTriggerDispatcherMetricCount(t *testing.T) {
broker, _, _, _ := newEventBrokerForTest()
defer broker.close()

info := newMockDispatcherInfo(t, 100, common.NewDispatcherID(), common.DDLSpanTableID, eventpb.ActionType_ACTION_TYPE_REGISTER)
info.span = common.KeyspaceDDLSpan(testTableTriggerKeyspaceID)

baseline := testutil.ToFloat64(metrics.EventServiceDispatcherGauge.WithLabelValues("1"))
require.NoError(t, broker.addDispatcher(info))
require.InDelta(t, baseline+1, testutil.ToFloat64(metrics.EventServiceDispatcherGauge.WithLabelValues("1")), 1e-9)

broker.removeDispatcher(info)
require.InDelta(t, baseline, testutil.ToFloat64(metrics.EventServiceDispatcherGauge.WithLabelValues("1")), 1e-9)
}

func TestScanRangeCappedByScanWindow(t *testing.T) {
broker, _, _, _ := newEventBrokerForTest()
// Close the broker, so we can catch all message in the test.
Expand Down
2 changes: 2 additions & 0 deletions pkg/eventservice/event_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,7 @@ func (s *eventService) handleMessage(ctx context.Context, msg *messaging.TargetM
case messaging.TypeDispatcherHeartbeat:
if len(msg.Message) != 1 {
log.Warn("invalid dispatcher heartbeat, ignore it", zap.Any("msg", msg))
return nil
Comment thread
lidezhu marked this conversation as resolved.
}
heartbeat := msg.Message[0].(*event.DispatcherHeartbeat)
select {
Expand All @@ -179,6 +180,7 @@ func (s *eventService) handleMessage(ctx context.Context, msg *messaging.TargetM
case messaging.TypeCongestionControl:
if len(msg.Message) != 1 {
log.Warn("invalid control message, ignore it", zap.Any("msg", msg))
return nil
Comment thread
lidezhu marked this conversation as resolved.
}
m := msg.Message[0].(*event.CongestionControl)
s.handleCongestionControl(msg.From, m)
Expand Down
20 changes: 20 additions & 0 deletions pkg/eventservice/event_service_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,26 @@ func TestEventServiceBasic(t *testing.T) {
}
}

func TestHandleMessageIgnoresInvalidSingleMessagePayloads(t *testing.T) {
es := &eventService{}

require.NotPanics(t, func() {
err := es.handleMessage(context.Background(), &messaging.TargetMessage{
Type: messaging.TypeDispatcherHeartbeat,
Message: nil,
})
require.NoError(t, err)
})

require.NotPanics(t, func() {
err := es.handleMessage(context.Background(), &messaging.TargetMessage{
Type: messaging.TypeCongestionControl,
Message: nil,
})
require.NoError(t, err)
})
}

var _ eventstore.EventStore = &mockEventStore{}

// mockEventStore is a mock implementation of the EventStore interface
Expand Down
4 changes: 4 additions & 0 deletions pkg/eventservice/test_helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ type mockSchemaStore struct {
maxDDLCommitTs uint64

registerTableError error
getTableInfoError error
}

func NewMockSchemaStore() *mockSchemaStore {
Expand Down Expand Up @@ -91,6 +92,9 @@ func (m *mockSchemaStore) SetTables(tables []commonEvent.Table) {
}

func (m *mockSchemaStore) GetTableInfo(keyspaceMeta common.KeyspaceMeta, tableID common.TableID, ts common.Ts) (*common.TableInfo, error) {
if m.getTableInfoError != nil {
return nil, m.getTableInfoError
}
if info, ok := m.TableInfo[tableID]; ok {
if info.deleteVersion <= uint64(ts) {
return nil, &schemastore.TableDeletedError{}
Expand Down
Loading