From ac0e78f07f1c2559e90fc2901a242c0878fd2d85 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Sun, 12 Jul 2026 08:24:38 +0800 Subject: [PATCH 1/2] eventservice: fix broker quota release and message validation --- pkg/eventservice/event_broker.go | 2 ++ pkg/eventservice/event_broker_test.go | 39 ++++++++++++++++++++++++++ pkg/eventservice/event_service.go | 2 ++ pkg/eventservice/event_service_test.go | 20 +++++++++++++ 4 files changed, 63 insertions(+) diff --git a/pkg/eventservice/event_broker.go b/pkg/eventservice/event_broker.go index 4e0a659f05..da274c2503 100644 --- a/pkg/eventservice/event_broker.go +++ b/pkg/eventservice/event_broker.go @@ -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)) 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() @@ -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), diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go index 96117255e1..34c59ec732 100644 --- a/pkg/eventservice/event_broker_test.go +++ b/pkg/eventservice/event_broker_test.go @@ -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" @@ -204,6 +206,43 @@ 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 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. diff --git a/pkg/eventservice/event_service.go b/pkg/eventservice/event_service.go index 8839ff9d87..345654e12a 100644 --- a/pkg/eventservice/event_service.go +++ b/pkg/eventservice/event_service.go @@ -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 } heartbeat := msg.Message[0].(*event.DispatcherHeartbeat) select { @@ -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 } m := msg.Message[0].(*event.CongestionControl) s.handleCongestionControl(msg.From, m) diff --git a/pkg/eventservice/event_service_test.go b/pkg/eventservice/event_service_test.go index 166df9429e..d4aae5c4e4 100644 --- a/pkg/eventservice/event_service_test.go +++ b/pkg/eventservice/event_service_test.go @@ -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 From 47a25dcf0af3438db3eec0510875f0902cb6f70e Mon Sep 17 00:00:00 2001 From: lidezhu Date: Sun, 12 Jul 2026 09:47:36 +0800 Subject: [PATCH 2/2] fix --- pkg/eventservice/event_broker.go | 12 +++++----- pkg/eventservice/event_broker_test.go | 32 +++++++++++++++++++++++++++ pkg/eventservice/test_helper.go | 4 ++++ 3 files changed, 42 insertions(+), 6 deletions(-) diff --git a/pkg/eventservice/event_broker.go b/pkg/eventservice/event_broker.go index da274c2503..437fbf0ec3 100644 --- a/pkg/eventservice/event_broker.go +++ b/pkg/eventservice/event_broker.go @@ -715,17 +715,12 @@ 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()), @@ -733,6 +728,11 @@ func (c *eventBroker) doScan(ctx context.Context, task scanTask) { 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))) diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go index 34c59ec732..440bdbd344 100644 --- a/pkg/eventservice/event_broker_test.go +++ b/pkg/eventservice/event_broker_test.go @@ -228,6 +228,38 @@ func TestDoScanReleasesChangefeedQuotaOnDispatcherQuotaFailure(t *testing.T) { 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() diff --git a/pkg/eventservice/test_helper.go b/pkg/eventservice/test_helper.go index 06727c2e96..5ca729eee9 100644 --- a/pkg/eventservice/test_helper.go +++ b/pkg/eventservice/test_helper.go @@ -39,6 +39,7 @@ type mockSchemaStore struct { maxDDLCommitTs uint64 registerTableError error + getTableInfoError error } func NewMockSchemaStore() *mockSchemaStore { @@ -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{}