From 1e085bcf6c2b004301d7c2f6b57decbc53f2399b Mon Sep 17 00:00:00 2001 From: tharanga Date: Thu, 4 Jun 2026 06:49:17 -0700 Subject: [PATCH 1/3] server: skip campaign until region syncer history phase is done Signed-off-by: tharanga re-pro of a new PD node acquiring the leadership with no regions fixing the test to fail without any changes Signed-off-by: tharanga server: skip campaign until region syncer history phase done Signed-off-by: tharanga --- pkg/storage/endpoint/region_syncer.go | 54 +++++++++ pkg/storage/endpoint/region_syncer_test.go | 77 ++++++++++++ pkg/storage/storage.go | 1 + pkg/syncer/client.go | 53 +++++++- pkg/syncer/client_test.go | 50 ++++++++ pkg/syncer/history_buffer.go | 42 +++++++ pkg/syncer/history_buffer_test.go | 74 ++++++++++++ pkg/syncer/server.go | 83 +++++++++++-- pkg/syncer/server_test.go | 23 +++- pkg/utils/keypath/absolute_key_path.go | 10 ++ server/server.go | 100 +++++++++++++++ .../region_syncer/region_syncer_test.go | 114 ++++++++++++++++++ 12 files changed, 667 insertions(+), 14 deletions(-) create mode 100644 pkg/storage/endpoint/region_syncer.go create mode 100644 pkg/storage/endpoint/region_syncer_test.go diff --git a/pkg/storage/endpoint/region_syncer.go b/pkg/storage/endpoint/region_syncer.go new file mode 100644 index 00000000000..a9b48c79cfc --- /dev/null +++ b/pkg/storage/endpoint/region_syncer.go @@ -0,0 +1,54 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package endpoint + +import ( + "strconv" + + "github.com/tikv/pd/pkg/errs" + "github.com/tikv/pd/pkg/utils/keypath" +) + +// RegionSyncerStorage defines the storage operations on the region syncer's +// cluster-level committed state. +type RegionSyncerStorage interface { + LoadRegionSyncerCommittedRegionCount() (uint64, error) + SaveRegionSyncerCommittedRegionCount(count uint64) error +} + +var _ RegionSyncerStorage = (*StorageEndpoint)(nil) + +// LoadRegionSyncerCommittedRegionCount loads the region count the current +// leader last published. It returns (0, nil) when the key is absent (a fresh +// cluster or one upgraded from a version that never wrote it), which callers +// treat as "no committed regions". +func (se *StorageEndpoint) LoadRegionSyncerCommittedRegionCount() (uint64, error) { + value, err := se.Load(keypath.RegionSyncerCommittedRegionCountPath()) + if err != nil || value == "" { + return 0, err + } + count, err := strconv.ParseUint(value, 10, 64) + if err != nil { + return 0, errs.ErrStrconvParseUint.Wrap(err).GenWithStackByArgs() + } + return count, nil +} + +// SaveRegionSyncerCommittedRegionCount persists the region count the current +// leader is serving so other members can tell whether they are caught up +// before campaigning for PD leadership. +func (se *StorageEndpoint) SaveRegionSyncerCommittedRegionCount(count uint64) error { + return se.Save(keypath.RegionSyncerCommittedRegionCountPath(), strconv.FormatUint(count, 10)) +} diff --git a/pkg/storage/endpoint/region_syncer_test.go b/pkg/storage/endpoint/region_syncer_test.go new file mode 100644 index 00000000000..0221ba572ab --- /dev/null +++ b/pkg/storage/endpoint/region_syncer_test.go @@ -0,0 +1,77 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package endpoint + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/tikv/pd/pkg/storage/kv" + "github.com/tikv/pd/pkg/utils/keypath" +) + +// newMemStorageEndpoint returns a StorageEndpoint backed by an in-memory KV, +// for tests that exercise endpoint encode/decode round-trips without etcd. +func newMemStorageEndpoint() *StorageEndpoint { + return NewStorageEndpoint(kv.NewMemoryKV(), nil) +} + +// TestRegionSyncerCommittedRegionCountAbsent verifies the back-compat path the +// leader-election gate relies on: an unwritten key (fresh or pre-upgrade +// cluster) must read back as (0, nil) rather than an error, so the gate treats +// it as "no committed regions" and allows the campaign. +func TestRegionSyncerCommittedRegionCountAbsent(t *testing.T) { + re := require.New(t) + se := newMemStorageEndpoint() + + count, err := se.LoadRegionSyncerCommittedRegionCount() + re.NoError(err) + re.Equal(uint64(0), count) +} + +// TestRegionSyncerCommittedRegionCountRoundTrip verifies Save/Load preserves the +// value across the uint64<->string boundary, including the zero (empty cluster) +// and max-uint64 edges, and that a later Save overwrites the prior value. +func TestRegionSyncerCommittedRegionCountRoundTrip(t *testing.T) { + re := require.New(t) + se := newMemStorageEndpoint() + + for _, want := range []uint64{0, 1, 110, math.MaxUint64} { + re.NoError(se.SaveRegionSyncerCommittedRegionCount(want)) + got, err := se.LoadRegionSyncerCommittedRegionCount() + re.NoError(err) + re.Equal(want, got) + } + + // A subsequent write replaces the prior value (counts shrink on merges). + re.NoError(se.SaveRegionSyncerCommittedRegionCount(42)) + got, err := se.LoadRegionSyncerCommittedRegionCount() + re.NoError(err) + re.Equal(uint64(42), got) +} + +// TestRegionSyncerCommittedRegionCountCorruptValue verifies a non-numeric value +// (corruption or a manual edit) surfaces as an error so the gate falls back to +// allowing the campaign rather than silently treating it as zero. +func TestRegionSyncerCommittedRegionCountCorruptValue(t *testing.T) { + re := require.New(t) + se := newMemStorageEndpoint() + + re.NoError(se.Save(keypath.RegionSyncerCommittedRegionCountPath(), "not-a-number")) + _, err := se.LoadRegionSyncerCommittedRegionCount() + re.Error(err) +} diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 7ee13b845b2..02dcce52a87 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -43,6 +43,7 @@ type Storage interface { endpoint.GCStateStorage endpoint.MinResolvedTSStorage endpoint.ExternalTSStorage + endpoint.RegionSyncerStorage endpoint.KeyspaceStorage endpoint.ResourceGroupStorage endpoint.TSOStorage diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index e6282006513..c3bcfcdac34 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -53,6 +53,14 @@ const ( func (s *RegionSyncer) StopSyncWithLeader() { s.reset() s.wg.Wait() + // If this sync session never observed an end-of-history marker, the + // in-memory history index reflects a partial transfer that the leader + // never confirmed. Roll back to the last committed (persisted) index + // so the next leader sees a StartIndex that triggers a fresh bulk + // rather than a stale offset into a different leader's history space. + if !s.historySynced.Load() { + s.history.rollback() + } } func (s *RegionSyncer) reset() { @@ -70,6 +78,22 @@ func (s *RegionSyncer) ResetHistoryIndex(index uint64) { s.history.resetWithIndexAndPersist(index) } +// HasAttemptedSync reports whether StartSyncWithLeader has ever been called +// during this process's lifetime. Used by the leader-election path to tell +// "I am a fresh single-node cluster" apart from "I am a follower that needs +// to be caught up before campaigning". +func (s *RegionSyncer) HasAttemptedSync() bool { + return s.attemptedSync.Load() +} + +// IsHistorySynced reports whether this server has, at some point during its +// lifetime, observed that it was caught up to a leader's history. The signal +// is sticky: once true, the local region storage is durably populated and +// further syncs only extend that state. +func (s *RegionSyncer) IsHistorySynced() bool { + return s.historySynced.Load() +} + func (s *RegionSyncer) syncRegion(ctx context.Context, conn *grpc.ClientConn) (ClientStream, error) { cli := pdpb.NewPDClient(conn) syncStream, err := cli.SyncRegions(ctx) @@ -126,6 +150,22 @@ func (s *RegionSyncer) handleRegionSyncResponse( } hasStats := len(stats) == len(regions) hasBuckets := len(buckets) == len(regions) + // An empty-regions response is the explicit end-of-history + // marker: sent by the leader at the end of a bulk transfer, + // at the end of an incremental catch-up, as an "already in + // sync" reply, or as a keepalive. Receiving one commits the + // historical phase — we flush the accumulated history index + // to disk and flip historySynced, which both opens the + // leader-election gate and switches subsequent records from + // the non-persisting catch-up path to the normal persisting + // live-stream path. + if len(regions) == 0 { + if !s.historySynced.Load() { + s.history.commit() + } + s.historySynced.Store(true) + } + inCatchup := !s.historySynced.Load() for i, r := range regions { var ( region *core.RegionInfo @@ -165,7 +205,16 @@ func (s *RegionSyncer) handleRegionSyncResponse( err = regionStorage.SaveRegion(r) } if err == nil && !inFullSync { - s.history.record(region) + // Full-sync frames carry positional offsets, not reusable history + // indices, so they are applied to storage but not recorded here. + // Records during historical catch-up are buffered without persisting + // and flushed by commit() on the end-of-history marker; live records + // after catch-up persist normally. + if inCatchup { + s.history.recordNoPersist(region) + } else { + s.history.record(region) + } } for _, old := range overlaps { _ = regionStorage.DeleteRegion(old.GetMeta()) @@ -187,7 +236,7 @@ func (s *RegionSyncer) IsRunning() bool { // StartSyncWithLeader starts to sync with leader. func (s *RegionSyncer) StartSyncWithLeader(addr string) { s.wg.Add(1) - + s.attemptedSync.Store(true) s.mu.Lock() defer s.mu.Unlock() s.mu.clientCtx, s.mu.clientCancel = context.WithCancel(s.server.LoopContext()) diff --git a/pkg/syncer/client_test.go b/pkg/syncer/client_test.go index 0591b7259e0..284c538f366 100644 --- a/pkg/syncer/client_test.go +++ b/pkg/syncer/client_test.go @@ -16,6 +16,7 @@ package syncer import ( "context" + "strconv" "testing" "time" @@ -31,6 +32,7 @@ import ( "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/mock/mockserver" "github.com/tikv/pd/pkg/storage" + "github.com/tikv/pd/pkg/storage/kv" "github.com/tikv/pd/pkg/utils/grpcutil" "github.com/tikv/pd/pkg/utils/keypath" "github.com/tikv/pd/pkg/utils/testutil" @@ -73,6 +75,54 @@ func TestLoadRegion(t *testing.T) { re.Less(time.Since(start), time.Second*2) } +// TestHistorySyncedInitFromDurableState verifies that NewRegionSyncer +// seeds historySynced from the persisted history index so a node that +// previously completed a sync can still campaign after a restart followed +// by a leader death mid-sync. A fresh-on-disk node must stay false so it +// is forced through a real catch-up before it can campaign. +func TestHistorySyncedInitFromDurableState(t *testing.T) { + re := require.New(t) + + newSyncer := func(seed func(kv.Base)) *RegionSyncer { + tempDir := t.TempDir() + rs, err := storage.NewRegionStorageWithLevelDBBackend(context.Background(), tempDir, nil) + re.NoError(err) + t.Cleanup(func() { re.NoError(rs.Close()) }) + seed(rs) + server := mockserver.NewMockServer( + context.Background(), + nil, + nil, + storage.NewCoreStorage(storage.NewStorageWithMemoryBackend(), rs), + core.NewBasicCluster(), + ) + return NewRegionSyncer(server) + } + + // Fresh KV: no persisted historyIndex. Stays false so the node is + // gated until it actually catches up to a leader. + rc := newSyncer(func(kv.Base) {}) + re.False(rc.IsHistorySynced(), + "fresh KV should not be treated as already synced") + + // Persisted historyIndex from a prior successful sync. The index only + // ever lands on disk after SaveRegion calls succeeded (via commit() or + // record()'s flushCount path), so a non-zero index is sufficient + // evidence of durable region state. + rc = newSyncer(func(b kv.Base) { + re.NoError(b.Save(historyKey, "42")) + }) + re.True(rc.IsHistorySynced(), + "persisted history index must initialize historySynced to true") + re.Equal(uint64(42), rc.history.getNextIndex(), + "history index should reload the persisted value") + // Sanity: a value that round-trips through strconv matches what + // history_buffer's persist path writes. + v, err := strconv.ParseUint("42", 10, 64) + re.NoError(err) + re.Equal(rc.history.getNextIndex(), v) +} + func TestErrorCode(t *testing.T) { re := require.New(t) tempDir := t.TempDir() diff --git a/pkg/syncer/history_buffer.go b/pkg/syncer/history_buffer.go index 7872aa1456e..afc332403e4 100644 --- a/pkg/syncer/history_buffer.go +++ b/pkg/syncer/history_buffer.go @@ -142,6 +142,23 @@ func (h *historyBuffer) record(r *core.RegionInfo) { } } +// recordNoPersist is like record but does not advance the on-disk +// historyIndex. The syncer client uses it during a historical catch-up +// so that a partially-received bulk does not commit a non-zero index; if +// the catch-up is interrupted (leader failover, process crash), the next +// sync starts from the last committed index instead of the middle of a +// half-applied transfer. +func (h *historyBuffer) recordNoPersist(r *core.RegionInfo) { + h.Lock() + defer h.Unlock() + syncIndexGauge.Set(float64(h.index)) + h.records[h.tail] = r + h.tail = (h.tail + 1) % h.size + if h.tail == h.head { + h.head = (h.head + 1) % h.size + } + h.index++ +} func (h *historyBuffer) recordsFrom(index uint64) []*core.RegionInfo { h.RLock() defer h.RUnlock() @@ -386,3 +403,28 @@ func (h *historyBuffer) getLocked(index uint64) *core.RegionInfo { } return nil } + +// commit persists the current historyIndex. The syncer client calls this +// when an end-of-history marker is received, to atomically commit a bulk +// or catch-up transfer. +func (h *historyBuffer) commit() { + h.Lock() + defer h.Unlock() + h.persist() + h.flushCount = defaultFlushCount +} + +// rollback resets the in-memory historyIndex back to the last value +// persisted on disk and clears the ring buffer. The syncer client calls +// this when a sync session ends without ever receiving an end-of-history +// marker, so the next sync starts from the last committed index (0 for a +// fresh member that never completed a catch-up). +func (h *historyBuffer) rollback() { + h.Lock() + defer h.Unlock() + h.index = 0 + h.head = 0 + h.tail = 0 + h.flushCount = defaultFlushCount + h.reload() +} diff --git a/pkg/syncer/history_buffer_test.go b/pkg/syncer/history_buffer_test.go index 83a28a0eb59..271917b53d5 100644 --- a/pkg/syncer/history_buffer_test.go +++ b/pkg/syncer/history_buffer_test.go @@ -193,3 +193,77 @@ func newTestHistoryBuffer(maxCapacity int) *historyBuffer { func newHistoryBufferTestRegion(regionID uint64) *core.RegionInfo { return core.NewRegionInfo(&metapb.Region{Id: regionID}, nil) } + +// TestCatchupCommitRollback exercises the catch-up gating semantics added +// for the leader-election gate: a half-applied catch-up that is cut short +// must not leave a non-zero persisted index behind, and a clean end-of- +// history must commit the new index atomically. +func TestCatchupCommitRollback(t *testing.T) { + re := require.New(t) + var regions []*core.RegionInfo + for i := range 10 { + regions = append(regions, core.NewRegionInfo(&metapb.Region{Id: uint64(i + 1)}, nil)) + } + + // Fresh member: nothing persisted, in-memory index starts at 0. + kvMem := kv.NewMemoryKV() + h := newHistoryBufferWithConfig(100, 100, 1, kvMem) + re.Equal(uint64(0), h.nextIndex()) + s, err := kvMem.Load(historyKey) + re.NoError(err) + re.Empty(s) + + // Apply a catch-up batch without persisting. In-memory index advances + // but the on-disk index must stay at 0. + for _, r := range regions[:5] { + h.recordNoPersist(r) + } + re.Equal(uint64(5), h.nextIndex()) + s, err = kvMem.Load(historyKey) + re.NoError(err) + re.Empty(s) + + // Simulate leader failover before end-of-history: rollback must wipe + // the buffered records and restore the in-memory index to the last + // persisted value (0 here). + h.rollback() + re.Equal(uint64(0), h.nextIndex()) + re.Equal(0, h.len()) + re.Nil(h.get(0)) + re.Nil(h.get(4)) + + // Re-attempt the catch-up against a new leader. This time we receive + // the end-of-history marker, so commit() persists the index. + for _, r := range regions[:5] { + h.recordNoPersist(r) + } + h.commit() + re.Equal(uint64(5), h.nextIndex()) + s, err = kvMem.Load(historyKey) + re.NoError(err) + re.Equal("5", s) + + // After a successful commit, a subsequent rollback (e.g., the next + // sync session ends without its own end-of-history marker before any + // further records arrive) must keep the committed index intact. + h.rollback() + re.Equal(uint64(5), h.nextIndex()) + s, err = kvMem.Load(historyKey) + re.NoError(err) + re.Equal("5", s) + + // A fresh buffer built against the same KV must see the committed + // index — this is the StartIndex the next sync session would send. + h2 := newHistoryBufferWithConfig(100, 100, 1, kvMem) + re.Equal(uint64(5), h2.nextIndex()) + + // Records appended after the committed point and rolled back again + // must not leak past the committed index. + for _, r := range regions[5:] { + h2.recordNoPersist(r) + } + re.Equal(uint64(10), h2.nextIndex()) + h2.rollback() + re.Equal(uint64(5), h2.nextIndex()) + re.Equal(0, h2.len()) +} diff --git a/pkg/syncer/server.go b/pkg/syncer/server.go index cf2c064c87f..c2b60e97078 100644 --- a/pkg/syncer/server.go +++ b/pkg/syncer/server.go @@ -50,9 +50,13 @@ const ( maxSyncRegionBatchSize = 1000 syncerKeepAliveInterval = 10 * time.Second historyBufferShrinkInterval = 5 * time.Minute - defaultHistoryBufferSize = 10000 - historyBufferMemoryStep = 4 * 1024 * 1024 * 1024 - maxHistoryBufferSize = 80000 + // The frequency to check and publish the region count the leader is + // serving so that other members can tell whether they are caught up + // before campaigning. + committedRegionCountInterval = time.Second + defaultHistoryBufferSize = 10000 + historyBufferMemoryStep = 4 * 1024 * 1024 * 1024 + maxHistoryBufferSize = 80000 ) // ClientStream is the client side of the region syncer. @@ -193,6 +197,13 @@ type RegionSyncer struct { tlsConfig *grpcutil.TLSConfig // status when as client streamingRunning atomic.Bool + // attempted sync status as client, sticky for the process lifetime. + // It is used to distinguish follower from never attempted to sync (e.g., bootstrapp). + attemptedSync atomic.Bool + // status of the historitcal catch-up as client, sticky for the process lifetime. + // set to true once the client has observed that it has completed the historitcal + // catch-up/ the local region storage is durably populated. + historySynced atomic.Bool } // NewRegionSyncer returns a region syncer that ensures final consistency through the heartbeat, @@ -211,6 +222,7 @@ func NewRegionSyncer(s Server) *RegionSyncer { tlsConfig: s.GetTLSConfig(), } syncer.mu.streams = make(map[string]*regionSyncStream) + syncer.reloadHistorySyncedFromDurableState() return syncer } @@ -229,12 +241,44 @@ func historyBufferMaxSizeFromMemory(totalMemory uint64) int { return size } +// This function seeds the in-memory historySynced +// flag from the persisted historyIndex so a node that previously completed +// a sync isn't permanently locked out of campaigning after a restart +// followed by a leader death mid-sync. A non-zero index only ever lands on +// disk via commit() (after a bulk applied via SaveRegion) or via record()'s +// flushCount path, so a non-zero index implies the local region storage was +// populated by a prior successful catch-up — sufficient evidence of durable +// state without any extra probe. Invoked once from NewRegionSyncer; safe to +// call before any sync session because nothing else has touched historySynced yet. +// TODO : this doesn't attempt to fix the gap problem https://github.com/tikv/pd/issues/10668 +func (s *RegionSyncer) reloadHistorySyncedFromDurableState() { + if s.history.getNextIndex() > 0 { + s.historySynced.Store(true) + } +} + // RunServer runs the server of the region syncer. // regionNotifier is used to get the changed regions. func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *core.RegionInfo) { var records []*core.RegionInfo keepAliveTicker := time.NewTicker(syncerKeepAliveInterval) shrinkTicker := time.NewTicker(historyBufferShrinkInterval) + committedCountTicker := time.NewTicker(committedRegionCountInterval) + // -1 forces an initial publish on the first tick. + lastPublishedRegionCount := -1 + publishCommittedRegionCount := func() { + if count := s.server.GetBasicCluster().GetTotalRegionCount(); count != lastPublishedRegionCount { + if err := s.server.GetStorage().SaveRegionSyncerCommittedRegionCount(uint64(count)); err != nil { + log.Warn("failed to persist committed region count", errs.ZapError(err)) + } else { + lastPublishedRegionCount = count + } + } + } + // Publish the region count immediately to minimize the poissibility of a leader dies + // immediately and a new campaign gate misclassifies an unsynced follower as + // already caught up. + publishCommittedRegionCount() processRegion := func(region *core.RegionInfo) { records = append(records, region) @@ -244,6 +288,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor defer func() { keepAliveTicker.Stop() shrinkTicker.Stop() + committedCountTicker.Stop() s.mu.Lock() for _, stream := range s.mu.streams { stream.close() @@ -274,6 +319,12 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor s.broadcast(ctx, records, false) case <-shrinkTicker.C: s.history.maybeShrink() + case <-committedCountTicker.C: + // Publish the region count this leader is serving so that a member + // that has not finished a history sync can still decide, after this + // leader is gone, whether it is caught up enough to campaign. Only + // the leader runs RunServer, so only the leader writes this. + publishCommittedRegionCount() case <-keepAliveTicker.C: s.broadcast(ctx, nil, true) } @@ -420,7 +471,7 @@ func (s *RegionSyncer) syncHistoryRegionLocked( zap.Uint64("from-index", startIndex), zap.Uint64("last-index", endIndex), zap.Int("records-length", len(records))) - return s.syncHistoryRecordsLocked(startIndex, records, syncStream) + return s.syncHistoryRecordsLocked(startIndex, records, syncStream, true) } func buildSyncRegionResponse(startIndex uint64, records []*core.RegionInfo) *pdpb.SyncRegionResponse { @@ -455,17 +506,28 @@ func buildSyncRegionResponse(startIndex uint64, records []*core.RegionInfo) *pdp func (s *RegionSyncer) syncHistoryRecords(startIndex uint64, records []*core.RegionInfo, stream *regionSyncStream) error { stream.sendMu.Lock() defer stream.sendMu.Unlock() - return s.syncHistoryRecordsLocked(startIndex, records, stream) + return s.syncHistoryRecordsLocked(startIndex, records, stream, true) } -func (*RegionSyncer) syncHistoryRecordsLocked(startIndex uint64, records []*core.RegionInfo, stream *regionSyncStream) error { +// syncHistoryRecordsLocked streams records in batches. When sendEndMarker is +// true it appends an end-of-history marker (empty regions) so the follower can +// commit its history index and flip historySynced. The full-sync path passes +// false because it streams positional batches and sends its own completion +// marker at the leader's real next index instead. +func (*RegionSyncer) syncHistoryRecordsLocked(startIndex uint64, records []*core.RegionInfo, stream *regionSyncStream, sendEndMarker bool) error { for start := 0; start < len(records); start += maxSyncRegionBatchSize { end := min(start+maxSyncRegionBatchSize, len(records)) if err := stream.sendStreamIfOpen(buildSyncRegionResponse(startIndex+uint64(start), records[start:end])); err != nil { return err } } - return nil + if !sendEndMarker { + return nil + } + return stream.sendStreamIfOpen(&pdpb.SyncRegionResponse{ + Header: &pdpb.ResponseHeader{ClusterId: keypath.ClusterID()}, + StartIndex: startIndex + uint64(len(records)), + }) } func (s *RegionSyncer) syncFullRegionsLocked(ctx context.Context, name string, syncStream *regionSyncStream, syncStartIndex uint64) error { @@ -542,11 +604,16 @@ func (s *RegionSyncer) syncFullRegionsLocked(ctx context.Context, name string, s if len(regions) == 0 { catchUpStartIndex = 0 } - if err := s.syncHistoryRecordsLocked(catchUpStartIndex, records, syncStream); err != nil { + // Full-sync batches are positional, so stream them without an + // end-of-history marker; the completion marker below carries the real + // next index. + if err := s.syncHistoryRecordsLocked(catchUpStartIndex, records, syncStream, false); err != nil { return err } syncStream.advanceSendIndexLocked(len(records)) } + // End-of-history marker at the leader's real next index so the follower + // commits the correct history index and flips historySynced. resp := &pdpb.SyncRegionResponse{ Header: &pdpb.ResponseHeader{ClusterId: keypath.ClusterID()}, StartIndex: nextIndex, diff --git a/pkg/syncer/server_test.go b/pkg/syncer/server_test.go index 3def6038330..7d1650adbc8 100644 --- a/pkg/syncer/server_test.go +++ b/pkg/syncer/server_test.go @@ -87,15 +87,21 @@ func TestSyncHistoryRecordsSplitBatches(t *testing.T) { } stream := newMockSyncRegionsServer() syncStream := newRegionSyncStream(stream, 10) - stream.sendCh = make(chan *pdpb.SyncRegionResponse, 2) + // Buffer the two record batches plus the trailing end-of-history marker so + // the synchronous syncHistoryRecords below never blocks on Send. + stream.sendCh = make(chan *pdpb.SyncRegionResponse, 3) re.NoError(syncer.syncHistoryRecords(10, records, syncStream)) first := <-stream.sendCh second := <-stream.sendCh + marker := <-stream.sendCh re.Equal(uint64(10), first.GetStartIndex()) re.Len(first.GetRegions(), maxSyncRegionBatchSize) re.Equal(uint64(10+maxSyncRegionBatchSize), second.GetStartIndex()) re.Len(second.GetRegions(), 1) + // The end-of-history marker carries the next index and no regions. + re.Empty(marker.GetRegions()) + re.Equal(uint64(10+len(records)), marker.GetStartIndex()) closedStream := &testServerStream{} closedSyncStream := newRegionSyncStream(closedStream, 10) @@ -222,8 +228,11 @@ func TestSyncFullRegionsKeepsLiveRecordsAppendedDuringCatchUp(t *testing.T) { re.NoError(syncer.sendDownstream(context.Background(), "pd-follower", syncStream, false)) responses := stream.sentResponses() re.Len(responses, 4) - re.Equal(liveStartIndex, responses[2].GetStartIndex()) + // responses[2] is the end-of-history marker emitted after the catch-up + // (empty regions, carrying the caught-up index). re.Empty(responses[2].GetRegions()) + re.Equal(liveStartIndex, responses[2].GetStartIndex()) + // responses[3] is the live record appended during catch-up, sent downstream. re.Equal(liveStartIndex, responses[3].GetStartIndex()) re.Equal([]*metapb.Region{{Id: 102}}, responses[3].GetRegions()) re.Equal(liveStartIndex+1, syncStream.getSendIndex()) @@ -810,8 +819,14 @@ func TestSyncHistoryRegionStopsAtSyncStartIndex(t *testing.T) { err := syncer.syncHistoryRegion(context.Background(), request, syncStream, syncStartIndex) re.NoError(err) - re.Equal(uint64(10), stream.lastResponse().GetStartIndex()) - re.Equal([]*metapb.Region{{Id: 1}}, stream.lastResponse().GetRegions()) + responses := stream.sentResponses() + re.Len(responses, 2) + // Replay stops before syncStartIndex: only region 1 (at index 10), not region 2. + re.Equal(uint64(10), responses[0].GetStartIndex()) + re.Equal([]*metapb.Region{{Id: 1}}, responses[0].GetRegions()) + // Followed by the end-of-history marker at syncStartIndex. + re.Empty(responses[1].GetRegions()) + re.Equal(syncStartIndex, responses[1].GetStartIndex()) } type testServerStream struct { diff --git a/pkg/utils/keypath/absolute_key_path.go b/pkg/utils/keypath/absolute_key_path.go index b3c9ecebe2e..5c149e29a3f 100644 --- a/pkg/utils/keypath/absolute_key_path.go +++ b/pkg/utils/keypath/absolute_key_path.go @@ -91,6 +91,10 @@ const ( storePathFormat = "/pd/%d/raft/s/%020d" // "/pd/{cluster_id}/raft/s/{store_id}" minResolvedTSPathFormat = "/pd/%d/raft/min_resolved_ts" // "/pd/{cluster_id}/raft/min_resolved_ts" externalTimestampPathFormat = "/pd/%d/raft/external_timestamp" // "/pd/{cluster_id}/raft/external_timestamp" + // regionSyncerCommittedRegionCountPathFormat holds the region count the + // current leader is serving, published so other members can tell whether + // they are caught up before campaigning for PD leadership. + regionSyncerCommittedRegionCountPathFormat = "/pd/%d/raft/region_syncer_committed_region_count" // "/pd/{cluster_id}/raft/region_syncer_committed_region_count" keyspaceMetaPrefixFormat = "/pd/%d/keyspaces/meta/" // "/pd/{cluster_id}/keyspaces/meta/" keyspaceMetaPathFormat = "/pd/%d/keyspaces/meta/%08d" // "/pd/{cluster_id}/keyspaces/meta/{keyspace_id}" @@ -204,6 +208,12 @@ func ExternalTimestampPath() string { return fmt.Sprintf(externalTimestampPathFormat, ClusterID()) } +// RegionSyncerCommittedRegionCountPath returns the path that stores the region +// count the current leader is serving. +func RegionSyncerCommittedRegionCountPath() string { + return fmt.Sprintf(regionSyncerCommittedRegionCountPathFormat, ClusterID()) +} + // RecoveringMarkPath returns the path to save the recovering mark. func RecoveringMarkPath() string { return fmt.Sprintf(recoveringMarkPathFormat, ClusterID()) diff --git a/server/server.go b/server/server.go index 6511e723cf6..d238fed5510 100644 --- a/server/server.go +++ b/server/server.go @@ -112,6 +112,16 @@ const ( lostPDLeaderMaxTimeoutSecs = 10 lostPDLeaderReElectionFactor = 10 + + // regionSyncerCampaignGrace bounds how long a member will refuse to campaign + // without a leader in the cluster because its region syncer has not caught up. + // This prevents a permanently leaderless cluster (the gate's circular dependency: + // a follower cannot finish syncing once the leader it would sync from is + // gone). + // It is set to the floor of the lost-PD-leader re-election timeout + // (randomTimeout in leaderLoop, whose minimum is lostPDLeaderMaxTimeoutSecs + // seconds plus lostPDLeaderReElectionFactor*ElectionInterval). + regionSyncerCampaignGrace = lostPDLeaderMaxTimeoutSecs * time.Second ) // EtcdStartTimeout the timeout of the startup etcd. @@ -1835,6 +1845,12 @@ func (s *Server) leaderLoop() { defer logutil.LogPanic() defer s.serverLoopWg.Done() + // gateBlockedSince records when this server first got blocked from + // campaigning by the region-syncer gate while leaderless; it bounds that + // wait via regionSyncerCampaignGrace. Reset whenever a leader exists or we + // proceed to campaign. + var gateBlockedSince time.Time + for { if s.IsClosed() { log.Info("server is closed, return PD leader loop") @@ -1854,6 +1870,9 @@ func (s *Server) leaderLoop() { continue } if leader != nil { + // A leader exists to (re)sync from, so the gate's leaderless grace + // no longer applies; reset it. + gateBlockedSince = time.Time{} err := s.reloadConfigFromKV() if err != nil { log.Error("reload config failed", errs.ZapError(err)) @@ -1905,10 +1924,91 @@ func (s *Server) leaderLoop() { time.Sleep(200 * time.Millisecond) continue } + // Refuse to campaign until the region syncer has caught up to the + // previous leader. Without this gate, a freshly joined follower + // whose syncer never produced a populated local store can win + // leadership immediately after the leader dies and serve a + // partial/empty region set until TiKV heartbeats repopulate it. + // `HasAttemptedSync` distinguishes "I was a follower" from "I + // started fresh and there is no leader to sync from" (the latter + // must be allowed to campaign so the cluster can bootstrap). + if !s.canCampaignAsRegionSyncerCaughtUp() { + if gateBlockedSince.IsZero() { + gateBlockedSince = time.Now() + } + if blocked := time.Since(gateBlockedSince); blocked < regionSyncerCampaignGrace { + log.Warn("skip campaigning of pd leader: region syncer has not caught up to the previous leader", + zap.String("server-name", s.Name()), + zap.Uint64("member-id", s.member.ID()), + zap.Duration("blocked-for", blocked)) + time.Sleep(200 * time.Millisecond) + continue + } + // Leaderless past the grace window with no caught-up member taking + // over: campaign anyway to restore availability rather than stay + // blocked forever. Any region gap on the new leader is repopulated + // by TiKV heartbeats. + log.Warn("region syncer has not caught up but campaign grace elapsed; campaigning to avoid a leaderless cluster", + zap.String("server-name", s.Name()), + zap.Uint64("member-id", s.member.ID()), + zap.Duration("grace", regionSyncerCampaignGrace)) + } + gateBlockedSince = time.Time{} s.campaignLeader() } } +// This function returns true when this server is eligible +// to campaign for PD leadership from a region-syncer correctness standpoint: +// - region storage is not in use (regions live in etcd, no syncer needed), or +// - the syncer was never started on this process (single-node bootstrap, or +// no leader has ever existed to sync from), or +// - the syncer has at some point observed that it was caught up to a +// leader's history (and thus the local region storage is durable), or +// - no leader has published any committed regions yet (a fresh or empty +// cluster, where there is nothing to be behind on). +// +// Otherwise this is a member that has been syncing from a leader that has +// region data but has not yet confirmed catch-up, so it must wait rather than +// win leadership with a stale/empty store ahead of its caught-up peers. +func (s *Server) canCampaignAsRegionSyncerCaughtUp() bool { + if !s.persistOptions.IsUseRegionStorage() { + return true + } + syncer := s.cluster.GetRegionSyncer() + if syncer == nil { + return true + } + if !syncer.HasAttemptedSync() { + return true + } + if syncer.IsHistorySynced() { + return true + } + // Not history-synced this session. Consult the durable region count the + // leader published so we neither deadlock the election nor let an empty + // member win. + committed, err := s.storage.LoadRegionSyncerCommittedRegionCount() + if err != nil { + // Missing state is already handled as committed == 0 by the storage + // endpoint. Any real error here means we cannot prove this member is + // safe to lead yet, so keep the gate closed. + log.Warn("failed to load committed region count, skipping campaign", + zap.String("server-name", s.Name()), errs.ZapError(err)) + return false + } + // committed == 0 means a fresh or empty cluster: nothing to be behind on. + // Note: the committed value could be reduced to a boolean "cluster has committed regions" + // flag. We keep the published region count for now until we are certain the magnitude is + // never needed (e.g. for a future tolerance/threshold policy). + if committed == 0 { + return true + } + // The cluster has committed regions and this member has not confirmed + // catch-up: it is behind, so it must not campaign yet. + return false +} + func (s *Server) campaignLeader() { log.Info("start to campaign PD leader", zap.String("campaign-leader-name", s.Name())) if err := s.member.Campaign(s.ctx, s.cfg.LeaderLease); err != nil { diff --git a/tests/server/region_syncer/region_syncer_test.go b/tests/server/region_syncer/region_syncer_test.go index 95bb6de1417..4b44da39743 100644 --- a/tests/server/region_syncer/region_syncer_test.go +++ b/tests/server/region_syncer/region_syncer_test.go @@ -426,3 +426,117 @@ func TestPrepareCheckerWithTransferLeader(t *testing.T) { re.True(rc.IsPrepared()) re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker")) } + +// TestUnsyncedMemberRefusesLeadership is a regression test for the race where +// a freshly joined PD with empty region storage could take over leadership +// before its region syncer had populated the local store. The fix gates the +// campaign in leaderLoop on the region syncer having observed catch-up with +// the previous leader. +// +// Test plan: +// 1. Bring up a 1-node cluster (pd1) with UseRegionStorage = true, populate +// it with regions, and wait for them to flush to leveldb. +// 2. Enable the `disableClientStreaming` failpoint so any new follower's +// syncer client fails to receive any regions — modelling a brand-new +// member that has not (yet) caught up. +// 3. Join pd2. It comes up as a follower but cannot sync. +// 4. Resign pd1's leadership. Without the fix, pd2 would campaign and win +// with an empty cache; with the fix, pd2 must refuse to campaign. +// 5. Poll for a window after the resign, asserting pd2 never holds +// leadership with a partial region set and that the syncer state +// machine stays in "attempted but not synced" — the precise state the +// gate is supposed to catch. +// +// Recovery (releasing the failpoint and letting pd2 eventually catch up and +// take leadership) is not exercised here because, in a 2-node cluster after +// `ResignLeader`, the cluster sits leaderless until the `leaderLoop`'s +// lost-PD-leader timeout (~40 s) fires and reclaims etcd leadership for +// pd1 — longer than a reasonable Eventually deadline. The gate's recovery +// behaviour is the same path the cluster takes on any natural leader +// failover and is covered by the existing TestRegionSyncer / TestFullSync* +// scenarios. +func TestUnsyncedMemberRefusesLeadership(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/storage/levelDBStorageFastFlush", `return(true)`)) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/storage/levelDBStorageFastFlush")) + }() + + cluster, err := tests.NewTestCluster(ctx, 1, func(conf *config.Config, _ string) { + conf.PDServerCfg.UseRegionStorage = true + }) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + re.NotEmpty(cluster.WaitLeader()) + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + rc := leaderServer.GetServer().GetRaftCluster() + re.NotNil(rc) + + // Populate the leader with regions and wait for them to be flushed to + // leveldb (the local region storage flush interval is ~3s). + regionLen := 110 + regions := tests.InitRegions(regionLen) + for _, region := range regions { + re.NoError(rc.HandleRegionHeartbeat(region)) + } + time.Sleep(4 * time.Second) + re.Len(leaderServer.GetServer().GetBasicCluster().GetRegions(), regionLen) + + // Block region-syncer clients from establishing a stream. Any follower + // that joins after this point will never receive history regions. + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/syncer/disableClientStreaming", `return(true)`)) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/syncer/disableClientStreaming")) + }() + + // Join a brand-new PD. Its leveldb starts empty and the failpoint + // prevents its syncer client from filling it. + pd2, err := cluster.Join(ctx) + re.NoError(err) + re.NoError(pd2.Run()) + re.Equal("pd1", cluster.WaitLeader()) + + // Confirm pd2 is wired up as a follower whose local region cache is still + // empty: the disableClientStreaming failpoint stops its syncer from ever + // establishing a stream, so it never catches up to pd1's history. + testutil.Eventually(re, func() bool { + return pd2.GetServer().DirectlyGetRaftCluster() != nil + }) + re.Empty(pd2.GetServer().GetBasicCluster().GetRegions()) + + // Resign pd1. Without the fix, pd2's leaderLoop would unblock from + // leader.Watch(), call StopSyncWithLeader, then campaign and win with an + // empty region cache. With the fix the gate forces pd2 to skip the + // campaign while it is not caught up. + re.NoError(cluster.ResignLeader()) + + // Observe the gate firing. pd2 must log "skip campaigning of pd leader: + // region syncer has not caught up" on every leader-loop iteration as long + // as the failpoint keeps its syncer from catching up. We poll for a few + // seconds to confirm the gate stays closed rather than flipping open by + // accident. + const observeWindow = 3 * time.Second + deadline := time.Now().Add(observeWindow) + for time.Now().Before(deadline) { + if pd2.IsLeader() && len(pd2.GetServer().GetBasicCluster().GetRegions()) < regionLen { + re.FailNowf("gate failed", + "unsynced pd2 became leader with %d/%d regions", + len(pd2.GetServer().GetBasicCluster().GetRegions()), regionLen) + } + time.Sleep(100 * time.Millisecond) + } + + // Final guard: pd2 must not hold leadership while its region cache is + // still empty. On unfixed code (no campaign gate) pd2 wins the election + // pd1 vacated and serves an empty region set, so this assertion — or the + // observe loop above — fails, reproducing the bug. + re.False(pd2.IsLeader(), "pd2 must not hold leadership while unsynced") + re.Empty(pd2.GetServer().GetBasicCluster().GetRegions(), + "pd2 region cache must stay empty while the syncer stream is blocked") +} From 007a7dd5952a27fb7144912d55f95914153dfe92 Mon Sep 17 00:00:00 2001 From: tharanga Date: Mon, 15 Jun 2026 14:40:24 -0700 Subject: [PATCH 2/3] fixing region syncer commit index to align with the syncer Signed-off-by: tharanga --- pkg/syncer/server.go | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/pkg/syncer/server.go b/pkg/syncer/server.go index c2b60e97078..a50c7d0a527 100644 --- a/pkg/syncer/server.go +++ b/pkg/syncer/server.go @@ -267,7 +267,19 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor // -1 forces an initial publish on the first tick. lastPublishedRegionCount := -1 publishCommittedRegionCount := func() { - if count := s.server.GetBasicCluster().GetTotalRegionCount(); count != lastPublishedRegionCount { + count := s.server.GetBasicCluster().GetTotalRegionCount() + // Only advertise a non-zero committed count once regions have actually + // entered the syncer history and thus become syncable by followers. + // Regions present in the basic cluster but that never flowed through the + // heartbeat -> changedRegions -> history path (most notably the bootstrap + // region) cannot have been synced by any follower. Counting them would + // keep the region-syncer campaign gate (canCampaignAsRegionSyncerCaughtUp) + // permanently closed and stall the next leader election until the grace + // elapses, even though no follower could possibly be caught up. + if s.history.getNextIndex() == 0 { + count = 0 + } + if count != lastPublishedRegionCount { if err := s.server.GetStorage().SaveRegionSyncerCommittedRegionCount(uint64(count)); err != nil { log.Warn("failed to persist committed region count", errs.ZapError(err)) } else { From 73f1b58549aae1af270025393d254d760a82066c Mon Sep 17 00:00:00 2001 From: tharanga Date: Fri, 19 Jun 2026 09:55:11 -0700 Subject: [PATCH 3/3] avoid transferring PD leadership to a member without region metadata Signed-off-by: tharanga --- pkg/member/member.go | 131 ++++++++++++++++++ pkg/member/member_test.go | 94 +++++++++++++ pkg/syncer/caught_up_members_test.go | 60 ++++++++ pkg/syncer/client.go | 40 ++++++ pkg/syncer/server.go | 31 ++++- server/server.go | 24 ++++ .../region_syncer/region_syncer_test.go | 92 ++++++++++++ 7 files changed, 469 insertions(+), 3 deletions(-) create mode 100644 pkg/member/member_test.go create mode 100644 pkg/syncer/caught_up_members_test.go diff --git a/pkg/member/member.go b/pkg/member/member.go index e8eaa52d296..eebc849492d 100644 --- a/pkg/member/member.go +++ b/pkg/member/member.go @@ -16,9 +16,11 @@ package member import ( "context" + "fmt" "math/rand/v2" "os" "path/filepath" + "slices" "strconv" "strings" "sync/atomic" @@ -64,6 +66,19 @@ type Member struct { memberValue string // lastLeaderUpdatedTime is the last time when the leader is updated. lastLeaderUpdatedTime atomic.Value + // caughtUpMembersFn, when set, returns the names of members whose region + // syncer has caught up to this leader's history. ResignEtcdLeader uses it + // to bias "transfer to any member" target selection toward a caught-up + // member, so the new leader is not stalled by the region-syncer campaign + // gate while its local region store is still empty. Stored as + // func() []string; nil when not configured. + caughtUpMembersFn atomic.Value + // hasCommittedRegionsFn, when set, reports whether the cluster has region + // data distributed through the syncer (committed > 0). ResignEtcdLeader uses + // it to decide whether refusing a not-caught-up transfer target is + // warranted: on a fresh/empty cluster (false) any member is a safe target. + // Stored as func() bool; nil when not configured. + hasCommittedRegionsFn atomic.Value } // NewMember create a new Member. @@ -85,6 +100,60 @@ func (m *Member) Name() string { return m.member.Name } +// SetCaughtUpMembersProvider registers a function that returns the names of +// members whose region syncer has caught up to this leader's history. It is +// consulted by ResignEtcdLeader to prefer a caught-up transfer target. Passing +// nil clears the provider. Safe to call concurrently with ResignEtcdLeader. +func (m *Member) SetCaughtUpMembersProvider(fn func() []string) { + if fn == nil { + m.caughtUpMembersFn.Store((func() []string)(nil)) + return + } + m.caughtUpMembersFn.Store(fn) +} + +// caughtUpMembers returns the configured caught-up member names, or nil when no +// provider is registered. +func (m *Member) caughtUpMembers() []string { + v := m.caughtUpMembersFn.Load() + if v == nil { + return nil + } + fn, ok := v.(func() []string) + if !ok || fn == nil { + return nil + } + return fn() +} + +// SetHasCommittedRegionsProvider registers a function reporting whether the +// cluster has region data distributed through the syncer. ResignEtcdLeader +// consults it before refusing a not-caught-up transfer target. Passing nil +// clears the provider. Safe to call concurrently with ResignEtcdLeader. +func (m *Member) SetHasCommittedRegionsProvider(fn func() bool) { + if fn == nil { + m.hasCommittedRegionsFn.Store((func() bool)(nil)) + return + } + m.hasCommittedRegionsFn.Store(fn) +} + +// hasCommittedRegions reports whether the cluster has syncable region data. It +// returns false when no provider is registered, so the transfer refusal is +// never triggered in setups that do not wire the signal (preserving the prior +// behavior of always honoring an explicit transfer target). +func (m *Member) hasCommittedRegions() bool { + v := m.hasCommittedRegionsFn.Load() + if v == nil { + return false + } + fn, ok := v.(func() bool) + if !ok || fn == nil { + return false + } + return fn() +} + // GetMember returns the member. func (m *Member) GetMember() any { return m.member @@ -367,18 +436,80 @@ func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLead return nil } + // Track candidate names alongside IDs so we can bias selection toward a + // region-syncer-caught-up member below. + candidateNames := make(map[uint64]string) for _, member := range res.Members { if (nextEtcdLeader == "" && member.ID != m.id) || (nextEtcdLeader != "" && member.Name == nextEtcdLeader) { etcdLeaderIDs = append(etcdLeaderIDs, member.GetID()) + candidateNames[member.GetID()] = member.GetName() } } if len(etcdLeaderIDs) == 0 { return errors.New("no valid pd to transfer etcd leader") } + if nextEtcdLeader == "" { + // When transferring to any member, prefer one whose region syncer has + // caught up to our history. Otherwise the new leader may stall the PD + // election behind the region-syncer campaign gate (and, because PD + // leadership follows etcd leadership, leave the cluster leaderless until + // the gate's grace elapses). Fall back to the full candidate set when no + // caught-up member is known, so resign never blocks on an empty set. + if preferred := filterCaughtUpCandidates(etcdLeaderIDs, candidateNames, m.caughtUpMembers()); len(preferred) > 0 { + log.Info("prefer region-syncer-caught-up members as etcd leader transfer targets", + zap.Int("candidates", len(etcdLeaderIDs)), zap.Int("caught-up", len(preferred))) + etcdLeaderIDs = preferred + } + } else if shouldRefuseTransferTarget(nextEtcdLeader, m.caughtUpMembers(), m.hasCommittedRegions()) { + // An explicit target that has not caught up would stall behind the + // campaign gate, leaving the cluster leaderless until the grace elapses. + // Refuse fast with a clear reason so the operator can pick a caught-up + // member or retry, instead of suffering a silent unavailability window. + // On a fresh/empty cluster (no syncable history) any member is safe, so + // this does not apply. + return fmt.Errorf("cannot transfer PD leadership to %q (choose a caught-up member or retry): %w", + nextEtcdLeader, ErrTransferTargetNotCaughtUp) + } nextEtcdLeaderID := etcdLeaderIDs[rand.IntN(len(etcdLeaderIDs))] return m.MoveEtcdLeader(ctx, m.ID(), nextEtcdLeaderID) } +// ErrTransferTargetNotCaughtUp is the cause returned when an explicit PD +// leadership transfer is refused because the target member's region syncer has +// not caught up to the current leader's history. +var ErrTransferTargetNotCaughtUp = errors.New("region syncer has not caught up to the current leader's history") + +// shouldRefuseTransferTarget reports whether an explicit transfer to target must +// be refused. It is refused only when the cluster has syncable region data +// (hasCommitted) and the target is not among the caught-up members; on a +// fresh/empty cluster any member is a safe target. +func shouldRefuseTransferTarget(target string, caughtUp []string, hasCommitted bool) bool { + if !hasCommitted { + return false + } + return !slices.Contains(caughtUp, target) +} + +// filterCaughtUpCandidates returns the subset of candidate IDs whose member +// name appears in caughtUp. It returns nil when caughtUp is empty or no +// candidate matches, signaling the caller to fall back to the full set. +func filterCaughtUpCandidates(candidateIDs []uint64, candidateNames map[uint64]string, caughtUp []string) []uint64 { + if len(caughtUp) == 0 { + return nil + } + caughtUpSet := make(map[string]struct{}, len(caughtUp)) + for _, name := range caughtUp { + caughtUpSet[name] = struct{}{} + } + var preferred []uint64 + for _, id := range candidateIDs { + if _, ok := caughtUpSet[candidateNames[id]]; ok { + preferred = append(preferred, id) + } + } + return preferred +} + // SetMemberLeaderPriority saves a member's priority to be elected as the etcd leader. func (m *Member) SetMemberLeaderPriority(id uint64, priority int) error { key := keypath.MemberLeaderPriorityPath(id) diff --git a/pkg/member/member_test.go b/pkg/member/member_test.go new file mode 100644 index 00000000000..a6905358e91 --- /dev/null +++ b/pkg/member/member_test.go @@ -0,0 +1,94 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package member + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/goleak" + + "github.com/tikv/pd/pkg/utils/testutil" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m, testutil.LeakOptions...) +} + +func TestFilterCaughtUpCandidates(t *testing.T) { + re := require.New(t) + names := map[uint64]string{1: "pd1", 2: "pd2", 3: "pd3"} + candidates := []uint64{1, 2, 3} + + // No caught-up info: fall back (nil) so the caller keeps the full set. + re.Nil(filterCaughtUpCandidates(candidates, names, nil)) + re.Nil(filterCaughtUpCandidates(candidates, names, []string{})) + + // Only the matching candidates are preferred; unknown names are ignored. + re.Equal([]uint64{2}, filterCaughtUpCandidates(candidates, names, []string{"pd2"})) + re.ElementsMatch([]uint64{1, 3}, filterCaughtUpCandidates(candidates, names, []string{"pd1", "pd3", "ghost"})) + + // A caught-up member that is not a valid transfer candidate yields nil so + // the caller falls back rather than transferring to a non-candidate. + re.Nil(filterCaughtUpCandidates([]uint64{1}, names, []string{"pd2", "pd3"})) +} + +func TestSetCaughtUpMembersProvider(t *testing.T) { + re := require.New(t) + m := &Member{} + + // Default: no provider configured. + re.Nil(m.caughtUpMembers()) + + m.SetCaughtUpMembersProvider(func() []string { return []string{"pd2"} }) + re.Equal([]string{"pd2"}, m.caughtUpMembers()) + + // Clearing the provider returns to the nil default without panicking. + m.SetCaughtUpMembersProvider(nil) + re.Nil(m.caughtUpMembers()) +} + +func TestSetHasCommittedRegionsProvider(t *testing.T) { + re := require.New(t) + m := &Member{} + + // Default (no provider): false, so the transfer refusal never triggers in + // setups that do not wire the signal. + re.False(m.hasCommittedRegions()) + + m.SetHasCommittedRegionsProvider(func() bool { return true }) + re.True(m.hasCommittedRegions()) + + m.SetHasCommittedRegionsProvider(func() bool { return false }) + re.False(m.hasCommittedRegions()) + + // Clearing the provider returns to the false default without panicking. + m.SetHasCommittedRegionsProvider(nil) + re.False(m.hasCommittedRegions()) +} + +func TestShouldRefuseTransferTarget(t *testing.T) { + re := require.New(t) + + // Empty/fresh cluster (no syncable history): any target is safe, never refuse. + re.False(shouldRefuseTransferTarget("pd2", nil, false)) + re.False(shouldRefuseTransferTarget("pd2", []string{}, false)) + re.False(shouldRefuseTransferTarget("pd2", []string{"pd3"}, false)) + + // Cluster has region data: refuse only when the target is not caught up. + re.False(shouldRefuseTransferTarget("pd2", []string{"pd2", "pd3"}, true)) + re.True(shouldRefuseTransferTarget("pd2", []string{"pd3"}, true)) + re.True(shouldRefuseTransferTarget("pd2", nil, true)) +} diff --git a/pkg/syncer/caught_up_members_test.go b/pkg/syncer/caught_up_members_test.go new file mode 100644 index 00000000000..ffca6270958 --- /dev/null +++ b/pkg/syncer/caught_up_members_test.go @@ -0,0 +1,60 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package syncer + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCaughtUpMembers(t *testing.T) { + re := require.New(t) + + // A leader with no bound streams reports nobody. + empty := &RegionSyncer{} + empty.mu.streams = map[string]*regionSyncStream{} + re.Nil(empty.CaughtUpMembers()) + + served := newRegionSyncStream(&testServerStream{}, 10) + served.markHistoryServed() + unserved := newRegionSyncStream(&testServerStream{}, 10) + + syncer := &RegionSyncer{} + syncer.mu.streams = map[string]*regionSyncStream{ + "pd-served": served, + "pd-unserved": unserved, + } + + // Only a stream that has finished historical catch-up is reported. + re.Equal([]string{"pd-served"}, syncer.CaughtUpMembers()) + + // Once the other stream completes its historical catch-up it joins. + unserved.markHistoryServed() + re.ElementsMatch([]string{"pd-served", "pd-unserved"}, syncer.CaughtUpMembers()) +} + +func TestHasSyncableHistory(t *testing.T) { + re := require.New(t) + + syncer := &RegionSyncer{history: newTestHistoryBuffer(8)} + // A fresh history buffer has distributed nothing. + re.False(syncer.HasSyncableHistory()) + + // Recording a region advances the history index, so there is now syncable + // history a follower could be behind on. + syncer.history.record(newHistoryBufferTestRegion(1)) + re.True(syncer.HasSyncableHistory()) +} diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index c3bcfcdac34..c173788741b 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -94,6 +94,46 @@ func (s *RegionSyncer) IsHistorySynced() bool { return s.historySynced.Load() } +// HasSyncableHistory reports whether this leader has distributed any region +// history through the syncer (its history buffer is non-empty). It mirrors the +// committed-region-count signal the campaign gate consults +// (canCampaignAsRegionSyncerCaughtUp): when false, a fresh or empty cluster has +// nothing for a follower to be behind on, so any member is a safe +// leadership-transfer target even if it is not "caught up". +func (s *RegionSyncer) HasSyncableHistory() bool { + return s.history.getNextIndex() > 0 +} + +// CaughtUpMembers returns the names of follower members whose region sync +// stream has completed its historical catch-up (the leader finished serving the +// bulk history and sent the end-of-history marker, so the follower transitions +// to the incremental live stream and flips its own historySynced). Such members +// are the safest targets for a leadership transfer: their local region store is +// populated, so they can satisfy the region-syncer campaign gate and take over +// immediately instead of stalling the election while empty. +// +// This is a best-effort hint: the per-stream historyServed flag can briefly +// lead the follower's own historySynced (the end-of-history marker is sent but +// not yet applied). That is safe because the target's own campaign gate +// (canCampaignAsRegionSyncerCaughtUp) remains the correctness backstop — a stale +// hint merely degrades target selection to the random fallback, never lets an +// empty member lead. Disconnected followers are unbound from the stream map, so +// they drop out of the candidate set. +// +// It is only meaningful on the current sync leader; on a follower the stream +// map is empty and it returns nil. +func (s *RegionSyncer) CaughtUpMembers() []string { + s.mu.RLock() + defer s.mu.RUnlock() + var names []string + for name, stream := range s.mu.streams { + if stream.historyHasBeenServed() { + names = append(names, name) + } + } + return names +} + func (s *RegionSyncer) syncRegion(ctx context.Context, conn *grpc.ClientConn) (ClientStream, error) { cli := pdpb.NewPDClient(conn) syncStream, err := cli.SyncRegions(ctx) diff --git a/pkg/syncer/server.go b/pkg/syncer/server.go index a50c7d0a527..5b36971018f 100644 --- a/pkg/syncer/server.go +++ b/pkg/syncer/server.go @@ -82,9 +82,30 @@ type regionSyncStream struct { syncutil.Mutex sendIndex uint64 } - notifyCh chan bool - done chan struct{} - once sync.Once + // historyServed flips true once the leader has finished serving this + // follower's historical catch-up (i.e. syncHistoryRegion returned and the + // end-of-history marker was sent), which is exactly when the follower + // transitions to the incremental live stream and flips its own + // historySynced. It is the leader-side mirror of that boundary, used by + // CaughtUpMembers to bias leadership-transfer target selection. It is a + // best-effort hint only: the target's own region-syncer campaign gate + // remains the correctness backstop. + historyServed atomic.Bool + notifyCh chan bool + done chan struct{} + once sync.Once +} + +// markHistoryServed records that this stream has completed the historical +// catch-up phase and entered incremental streaming. +func (s *regionSyncStream) markHistoryServed() { + s.historyServed.Store(true) +} + +// historyHasBeenServed reports whether the historical catch-up phase for this +// stream has completed. +func (s *regionSyncStream) historyHasBeenServed() bool { + return s.historyServed.Load() } func newRegionSyncStream(stream ServerStream, startIndex uint64) *regionSyncStream { @@ -388,6 +409,10 @@ func (s *RegionSyncer) Sync(ctx context.Context, stream pdpb.PD_SyncRegionsServe s.unbindStream(name, syncStream) return err } + // The historical catch-up is done and the end-of-history marker has been + // sent, so the follower is (about to be) caught up. Record it on the + // stream so CaughtUpMembers can prefer this follower as a transfer target. + syncStream.markHistoryServed() go s.runDownstreamSender(ctx, name, syncStream) select { case <-ctx.Done(): diff --git a/server/server.go b/server/server.go index d238fed5510..5bd3198a49a 100644 --- a/server/server.go +++ b/server/server.go @@ -533,6 +533,30 @@ func (s *Server) startServer(ctx context.Context) error { s.tsoAllocator = tso.NewAllocator(s.ctx, constant.DefaultKeyspaceGroupID, s.member, tsoStorage, s) s.basicCluster = core.NewBasicCluster() s.cluster = cluster.NewRaftCluster(ctx, s.GetMember(), s.GetBasicCluster(), s.GetStorage(), syncer.NewRegionSyncer(s), s.client, s.httpClient, s.tsoAllocator) + // Let leadership transfers prefer a region-syncer-caught-up target so the + // new leader is not stalled by the campaign gate (see + // canCampaignAsRegionSyncerCaughtUp). Resolve the syncer dynamically since + // the gate consults the live cluster too. + s.member.SetCaughtUpMembersProvider(func() []string { + if s.cluster == nil { + return nil + } + rs := s.cluster.GetRegionSyncer() + if rs == nil { + return nil + } + return rs.CaughtUpMembers() + }) + // Refuse an explicit leadership transfer to a not-caught-up target only when + // the cluster actually has syncable region data; on a fresh/empty cluster any + // member is a safe target. + s.member.SetHasCommittedRegionsProvider(func() bool { + if s.cluster == nil { + return false + } + rs := s.cluster.GetRegionSyncer() + return rs != nil && rs.HasSyncableHistory() + }) keyspaceIDAllocator := id.NewAllocator(&id.AllocatorParams{ Client: s.client, Label: id.KeyspaceLabel, diff --git a/tests/server/region_syncer/region_syncer_test.go b/tests/server/region_syncer/region_syncer_test.go index 4b44da39743..7157f5c498e 100644 --- a/tests/server/region_syncer/region_syncer_test.go +++ b/tests/server/region_syncer/region_syncer_test.go @@ -27,6 +27,7 @@ import ( "github.com/pingcap/failpoint" "github.com/tikv/pd/pkg/core" + "github.com/tikv/pd/pkg/member" "github.com/tikv/pd/pkg/utils/testutil" "github.com/tikv/pd/server/config" "github.com/tikv/pd/tests" @@ -540,3 +541,94 @@ func TestUnsyncedMemberRefusesLeadership(t *testing.T) { re.Empty(pd2.GetServer().GetBasicCluster().GetRegions(), "pd2 region cache must stay empty while the syncer stream is blocked") } + +// TestExplicitTransferRefusedForUnsyncedTarget verifies the ResignEtcdLeader +// decision for an explicit leadership transfer (e.g. the /leader/transfer/{name} +// API): when the cluster has region data and the target has not caught up, the +// transfer is refused with a clear reason and leadership does not move; on a +// fresh/empty cluster any target is allowed. The committed/caught-up signals are +// driven through the provider hooks so the decision is exercised deterministically +// against a real etcd MoveLeader, independently of sync timing. +func TestExplicitTransferRefusedForUnsyncedTarget(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cluster, err := tests.NewTestCluster(ctx, 2) + defer cluster.Destroy() + re.NoError(err) + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + target := cluster.GetFollower() + re.NotEmpty(target) + + m := cluster.GetLeaderServer().GetServer().GetMember() + + // Cluster has region data but the target has not caught up -> refuse fast + // with the typed reason; leadership must not move. + m.SetHasCommittedRegionsProvider(func() bool { return true }) + m.SetCaughtUpMembersProvider(func() []string { return nil }) + err = m.ResignEtcdLeader(ctx, leaderName, target) + re.ErrorIs(err, member.ErrTransferTargetNotCaughtUp) + re.ErrorContains(err, target) + re.Equal(leaderName, cluster.GetLeaderServer().GetServer().Name()) + + // Fresh/empty cluster (no syncable region data): any target is safe, so the + // same explicit transfer is allowed and leadership moves to the target. + m.SetHasCommittedRegionsProvider(func() bool { return false }) + re.NoError(m.ResignEtcdLeader(ctx, leaderName, target)) + testutil.Eventually(re, func() bool { + return cluster.GetServer(target).IsLeader() + }) +} + +// TestExplicitTransferAllowedForCaughtUpTarget verifies that an explicit +// transfer to a caught-up follower is allowed and moves leadership. It also +// exercises the real markHistoryServed -> CaughtUpMembers path on the leader. +func TestExplicitTransferAllowedForCaughtUpTarget(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/storage/levelDBStorageFastFlush", `return(true)`)) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/storage/levelDBStorageFastFlush")) + }() + + cluster, err := tests.NewTestCluster(ctx, 2, func(conf *config.Config, _ string) { + conf.PDServerCfg.UseRegionStorage = true + }) + defer cluster.Destroy() + re.NoError(err) + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + rc := leaderServer.GetServer().GetRaftCluster() + re.NotNil(rc) + + for _, region := range tests.InitRegions(50) { + re.NoError(rc.HandleRegionHeartbeat(region)) + } + re.True(rc.GetRegionSyncer().HasSyncableHistory()) + + // Wait until the follower is reported caught up in the leader's view; this + // is set by markHistoryServed once the leader finishes serving its history. + var target string + testutil.Eventually(re, func() bool { + caught := rc.GetRegionSyncer().CaughtUpMembers() + if len(caught) == 0 { + return false + } + target = caught[0] + return target != "" && target != leaderName + }) + + // Explicit transfer to a caught-up follower is allowed and moves leadership. + re.NoError(leaderServer.GetServer().GetMember().ResignEtcdLeader(ctx, leaderName, target)) + testutil.Eventually(re, func() bool { + return cluster.GetServer(target).IsLeader() + }) +}