diff --git a/pkg/mcs/resourcemanager/server/server.go b/pkg/mcs/resourcemanager/server/server.go index d041085255..f4e4a3b95a 100644 --- a/pkg/mcs/resourcemanager/server/server.go +++ b/pkg/mcs/resourcemanager/server/server.go @@ -214,10 +214,7 @@ func (s *Server) primaryElectionLoop() { // transfer target atomically and is cleaned up once this server wins. func (s *Server) campaignLeader(expectedPrimary string) bool { log.Info("start to campaign the primary/leader", zap.String("campaign-resource-manager-primary-name", s.participant.Name())) - var cmps []clientv3.Cmp - if cmp := utils.ExpectedPrimaryCmp(&s.participant.MsParam, expectedPrimary); cmp != nil { - cmps = append(cmps, *cmp) - } + cmps := []clientv3.Cmp{utils.ExpectedPrimaryCmp(&s.participant.MsParam, expectedPrimary)} if err := s.participant.CampaignWithCmps(s.Context(), s.cfg.LeaderLease, cmps...); err != nil { if err.Error() == errs.ErrEtcdTxnConflict.Error() { log.Info("campaign resource manager primary meets error due to txn conflict, another server may campaign successfully", @@ -245,8 +242,14 @@ func (s *Server) campaignLeader(expectedPrimary string) bool { // We have won the campaign, so the expected primary flag (if any) has served its // purpose as the affinity guard. Delete it so steady state is clean and a later - // failure re-elects immediately instead of waiting for the flag's TTL. - utils.DeleteExpectedPrimaryFlag(s.GetClient(), &s.participant.MsParam, expectedPrimary) + // failure re-elects immediately instead of waiting for the flag's TTL. If a newer + // transfer rewrote the flag to another member while we were winning, step down so + // the re-election routes leadership to that target. + if utils.DeleteExpectedPrimaryFlag(s.GetClient(), &s.participant.MsParam, expectedPrimary, s.participant) { + log.Info("the expected primary has been changed to another member, stepping down", + zap.String("server-name", s.Name())) + return false + } log.Info("triggering the primary callback functions") for _, cb := range s.primaryCallbacks { diff --git a/pkg/mcs/scheduling/server/server.go b/pkg/mcs/scheduling/server/server.go index ce621211a4..027c960640 100644 --- a/pkg/mcs/scheduling/server/server.go +++ b/pkg/mcs/scheduling/server/server.go @@ -287,10 +287,7 @@ func (s *Server) primaryElectionLoop() { // and to clean the flag up once this server wins. func (s *Server) campaignPrimary(expectedPrimary string) { log.Info("start to campaign the primary", zap.String("campaign-scheduling-primary-name", s.participant.Name())) - var cmps []clientv3.Cmp - if cmp := utils.ExpectedPrimaryCmp(&s.participant.MsParam, expectedPrimary); cmp != nil { - cmps = append(cmps, *cmp) - } + cmps := []clientv3.Cmp{utils.ExpectedPrimaryCmp(&s.participant.MsParam, expectedPrimary)} if err := s.participant.CampaignWithCmps(s.Context(), s.cfg.LeaderLease, cmps...); err != nil { if err.Error() == errs.ErrEtcdTxnConflict.Error() { log.Info("campaign scheduling primary meets error due to txn conflict, another server may campaign successfully", @@ -318,8 +315,14 @@ func (s *Server) campaignPrimary(expectedPrimary string) { // We have won the campaign, so the expected primary flag (if any) has served its // purpose as the affinity guard. Delete it so steady state is clean and a later - // failure re-elects immediately instead of waiting for the flag's TTL. - utils.DeleteExpectedPrimaryFlag(s.GetClient(), &s.participant.MsParam, expectedPrimary) + // failure re-elects immediately instead of waiting for the flag's TTL. If a newer + // transfer rewrote the flag to another member while we were winning, step down so + // the re-election routes leadership to that target. + if utils.DeleteExpectedPrimaryFlag(s.GetClient(), &s.participant.MsParam, expectedPrimary, s.participant) { + log.Info("the expected primary has been changed to another member, stepping down", + zap.String("server-name", s.Name())) + return + } log.Info("triggering the primary callback functions") for _, cb := range s.primaryCallbacks { diff --git a/pkg/mcs/utils/constant/constant.go b/pkg/mcs/utils/constant/constant.go index 0fd3ef64a0..2396b59339 100644 --- a/pkg/mcs/utils/constant/constant.go +++ b/pkg/mcs/utils/constant/constant.go @@ -44,7 +44,15 @@ const ( // campaign; if the target never comes up within this window the flag expires and // the cluster falls back to a free election. Tying it to the leader lease keeps the // window proportional to how fast leadership turns over. - TransferPrimaryLeaseMultiplier = int64(3) + // + // Kept at 1 (rather than a larger margin) on purpose: this multiplier directly + // bounds the cluster's worst-case unavailable window after a transfer whose target + // never wins a single campaign (down, unreachable, or stuck) - see TransferPrimary's + // doc comment. A bigger multiplier buys the target more slack but linearly extends + // that worst case; 1 lease already matches how long the cluster tolerates a primary + // being unreachable everywhere else (losing its own leader lease), so there is no + // reason for a transfer-induced outage to be allowed to run longer than that. + TransferPrimaryLeaseMultiplier = int64(1) // PrimaryTickInterval is the interval to check primary PrimaryTickInterval = 50 * time.Millisecond // LeaderTickInterval is the interval to check leader diff --git a/pkg/mcs/utils/expected_primary.go b/pkg/mcs/utils/expected_primary.go index 36d90dba80..94aacb11cc 100644 --- a/pkg/mcs/utils/expected_primary.go +++ b/pkg/mcs/utils/expected_primary.go @@ -56,12 +56,19 @@ type primaryData struct { output string } -// markExpectedPrimaryFlag marks the expected primary flag when the primary is specified. -func markExpectedPrimaryFlag(client *clientv3.Client, msParam *keypath.MsParam, primary *primaryData, leaseID clientv3.LeaseID) error { +// markExpectedPrimaryFlag marks the expected primary flag when the primary is +// specified. Extra cmps are folded into the same transaction as the Put, so a +// leader-key ownership guard built by the caller makes the marker write atomic +// with still holding leadership: if the caller lost leadership between its own +// IsServing() check and this call, the guard no longer holds and the Put is +// rejected instead of silently publishing a marker that the new, already-serving +// primary will never look at. +func markExpectedPrimaryFlag(client *clientv3.Client, msParam *keypath.MsParam, primary *primaryData, leaseID clientv3.LeaseID, cmps ...clientv3.Cmp) error { path := keypath.ExpectedPrimaryPath(msParam) log.Info("set expected primary flag", zap.String("primary-path", path), zap.String("primary", primary.output)) // write a flag to indicate the expected primary. resp, err := kv.NewSlowLogTxn(client). + If(cmps...). Then(clientv3.OpPut(path, primary.raw, clientv3.WithLease(leaseID))). Commit() if err != nil { @@ -75,81 +82,125 @@ func markExpectedPrimaryFlag(client *clientv3.Client, msParam *keypath.MsParam, return nil } -// DeleteExpectedPrimaryFlag deletes the expected primary flag once the target has -// won the campaign, so that in steady state the flag is absent and a subsequent -// failure of the new primary triggers a free re-election immediately instead of -// waiting for the flag's TTL to expire. -// -// The delete is conditional on the flag still holding `expectedValue` (the value -// the winner campaigned with). This prevents clobbering a newer transfer that may -// have already rewritten the flag to a different target while this primary was -// winning. It is best-effort: the flag's TTL is the backstop, so a failure here is -// only logged and never blocks serving. -// -// The flag is bound to an etcd lease, so after deleting the key we also revoke that -// lease. Otherwise we would leave the key gone but the lease alive — exactly the -// inconsistent state that issue #10875 is about — and leak a lease until its TTL -// expires. -func DeleteExpectedPrimaryFlag(client *clientv3.Client, msParam *keypath.MsParam, expectedValue string) { - if expectedValue == "" { - // Normal election without a transfer in progress, nothing to clean up. - return - } +// DeleteExpectedPrimaryFlag reconciles the expected primary flag after this member +// has won the campaign. In the common case the flag still holds expectedValue (or +// is already gone) and is deleted together with its lease, so steady state has no +// marker and later failures re-elect immediately instead of waiting for the marker +// TTL. The flag can also have been rewritten by a newer transfer while this member +// was winning; since a serving primary no longer watches the marker, that transfer +// would otherwise return success without ever taking effect. In that case: +// - if the newer marker still targets this member, it is deleted as well and the +// member keeps serving; +// - if it targets another member, DeleteExpectedPrimaryFlag returns true and the +// caller must step down so the re-election routes leadership to that target. +func DeleteExpectedPrimaryFlag(client *clientv3.Client, msParam *keypath.MsParam, expectedValue string, p *member.Participant) (superseded bool) { path := keypath.ExpectedPrimaryPath(msParam) - // Read the key (to capture its lease) and delete it in the same conditional txn. + current, deleted := deleteMarkerIfEquals(client, path, expectedValue) + if deleted { + log.Info("delete expected primary flag", zap.String("primary-path", path)) + return false + } + if current == "" { + // The marker is already gone (deleted, expired, or it never existed), or the + // transaction failed; in the latter case the marker TTL bounds the staleness. + return false + } + // The marker was rewritten by a newer transfer while this member was winning. + if p != nil && p.IsExpectedPrimary(current) { + // The newer transfer also targets this member, which is already serving. + // Clean the marker up so it does not linger until its TTL. Best effort: + // on failure the TTL applies. + if _, deleted := deleteMarkerIfEquals(client, path, current); deleted { + log.Info("delete expected primary flag rewritten by a newer transfer to this member", + zap.String("primary-path", path)) + } + return false + } + log.Info("expected primary flag was rewritten by a newer transfer while campaigning, step down", + zap.String("primary-path", path), zap.String("expected-value", expectedValue), + zap.String("current-value", current)) + return true +} + +// deleteMarkerIfEquals atomically deletes the expected primary marker and revokes +// its lease when its value equals want. It returns the marker value observed by the +// transaction ("" when the marker does not exist or the transaction failed) and +// whether the marker was deleted. Note that etcd value comparisons always fail on a +// missing key, so want == "" never deletes anything and reports the current value. +func deleteMarkerIfEquals(client *clientv3.Client, path, want string) (current string, deleted bool) { resp, err := kv.NewSlowLogTxn(client). - If(clientv3.Compare(clientv3.Value(path), "=", expectedValue)). + If(clientv3.Compare(clientv3.Value(path), "=", want)). Then(clientv3.OpGet(path), clientv3.OpDelete(path)). + Else(clientv3.OpGet(path)). Commit() if err != nil { log.Warn("failed to delete expected primary flag", zap.String("primary-path", path), errs.ZapError(err)) - return - } - if !resp.Succeeded { - log.Info("skip deleting expected primary flag, it has been changed or already gone", - zap.String("primary-path", path), zap.String("expected-value", expectedValue)) - return + return "", false } - log.Info("delete expected primary flag", zap.String("primary-path", path)) - // Revoke the lease the flag was bound to, if any, so no lease is leaked. kvs := resp.Responses[0].GetResponseRange().GetKvs() - if len(kvs) == 0 || kvs[0].Lease == 0 { - return + if !resp.Succeeded { + if len(kvs) == 0 { + return "", false + } + return string(kvs[0].Value), false } - leaseID := clientv3.LeaseID(kvs[0].Lease) - // Bound the revoke: this runs on the campaign path before the primary is promoted - // to serving, and the cleanup is best-effort, so a hung RPC must not block serving. - ctx, cancel := context.WithTimeout(client.Ctx(), etcdutil.DefaultRequestTimeout) - defer cancel() - if _, err := client.Revoke(ctx, leaseID); err != nil { - log.Warn("failed to revoke expected primary flag lease", - zap.String("primary-path", path), zap.Int64("lease-id", int64(leaseID)), errs.ZapError(err)) + // Deleted. Revoke the lease the marker was bound to, so the "key deleted but + // lease persists" state can never exist (#10875). + if len(kvs) > 0 && kvs[0].Lease != 0 { + leaseID := clientv3.LeaseID(kvs[0].Lease) + // Bound the revoke: this runs on the campaign path before the primary is + // promoted to serving, and the cleanup is best-effort, so a hung RPC must not + // block serving. + ctx, cancel := context.WithTimeout(client.Ctx(), etcdutil.DefaultRequestTimeout) + defer cancel() + if _, err := client.Revoke(ctx, leaseID); err != nil { + log.Warn("failed to revoke expected primary flag lease", + zap.String("primary-path", path), zap.Int64("lease-id", int64(leaseID)), errs.ZapError(err)) + } } + return want, true } -// ExpectedPrimaryCmp returns an etcd comparison asserting the expected primary flag -// still equals `expectedValue`. The caller appends it to the campaign transaction so -// that winning the leader key and "I am still the expected primary" become atomic: -// if a concurrent transfer rewrote the flag after it was read, the campaign txn -// fails and this member does not become primary. It returns nil when expectedValue -// is empty (normal election, no transfer in progress), in which case the campaign -// must not be constrained. -func ExpectedPrimaryCmp(msParam *keypath.MsParam, expectedValue string) *clientv3.Cmp { +// ExpectedPrimaryCmp returns an etcd comparison that guards a primary campaign +// against a concurrent transfer. +// - When expectedValue is non-empty, a transfer installed a marker naming this +// member as the target; the comparison asserts the marker still holds that +// value. +// - When expectedValue is empty, the campaigner observed no transfer in +// progress; the comparison asserts the marker is still absent. Without this +// a campaigner that paused after the read could resume after a transfer +// installed a target marker and released the leader key, then win the +// campaign and bypass the transfer affinity guard. +func ExpectedPrimaryCmp(msParam *keypath.MsParam, expectedValue string) clientv3.Cmp { + path := keypath.ExpectedPrimaryPath(msParam) if expectedValue == "" { - return nil + return clientv3.Compare(clientv3.CreateRevision(path), "=", 0) } - cmp := clientv3.Compare(clientv3.Value(keypath.ExpectedPrimaryPath(msParam)), "=", expectedValue) - return &cmp + return clientv3.Compare(clientv3.Value(path), "=", expectedValue) } // TransferPrimary transfers the primary of the specified service to a target member. // -// It writes the expected primary flag pointing at the target (with a TTL of a few -// leader leases, see constant.TransferPrimaryLeaseMultiplier) and then resigns the +// It writes the expected primary flag pointing at the target (with a TTL of +// constant.TransferPrimaryLeaseMultiplier leader leases) and then resigns the // current primary by revoking its leader lease, so the re-election picks up the -// target. The flag write -// happens before the resignation on purpose: it guarantees the affinity guard is in -// place before the leader key is released, so no other member can win the gap. +// target. The flag write happens before the resignation on purpose: it guarantees +// the affinity guard is in place before the leader key is released, so no other +// member can win the gap. +// +// TransferPrimary reports success as soon as the flag is written and the current +// primary has resigned - it does NOT wait for the target to actually win the +// campaign or finish initializing. The worst-case unavailability this can leave +// behind, if the target never wins a single campaign (down, unreachable, or stuck), +// is bounded by the flag's TTL: TransferPrimaryLeaseMultiplier * leaderLease, i.e. +// one leader lease. Once any member wins a campaign - the target or, after the TTL, +// any other live member via free election - the flag is cleared immediately +// (DeleteExpectedPrimaryFlag) regardless of whether that member goes on to +// initialize successfully, so a failure after winning does not extend this window; +// see DeleteExpectedPrimaryFlag and campaignPrimary's ordering. This is a deliberate +// trade-off: reporting success without waiting keeps the API call itself fast and +// independent of the caller's own timeout, at the cost of not confirming the +// transfer actually reached the requested target before returning. // // keyspaceGroupID is optional, only used for TSO service. p must be the participant // of the current serving primary (the API ensures the request runs on the primary). @@ -158,6 +209,26 @@ func TransferPrimary(client *clientv3.Client, p *member.Participant, serviceName if p == nil || !p.IsServing() { return errors.New("current member is not serving as primary, please check leadership") } + + // Capture the leader key's CreateRevision right after the IsServing() check, before + // discovery or the lease grant below - both are network round trips during which p + // could lose its lease and win a fresh campaign with the same MemberValue() (which + // never changes for the lifetime of the participant). CreateRevision changes on + // every fresh campaign (Leadership.Campaign requires CreateRevision(leaderKey) == 0 + // to win), so fencing the eventual marker write on the revision observed here - not + // on a value comparison, and not on a revision re-read later, closer to the write - + // ties the write to the exact leadership session IsServing() just checked for the + // whole duration of this function, not just its tail end. + leaderKeyPath := p.GetLeadership().GetLeaderKey() + leaderResp, err := client.Get(client.Ctx(), leaderKeyPath) + if err != nil { + return errors.Annotate(err, "failed to read leader key for transfer guard") + } + if len(leaderResp.Kvs) == 0 || string(leaderResp.Kvs[0].Value) != p.MemberValue() { + return errors.New("current member is not serving as primary, please check leadership") + } + leaderKeyGuard := clientv3.Compare(clientv3.CreateRevision(leaderKeyPath), "=", leaderResp.Kvs[0].CreateRevision) + log.Info("try to transfer primary", zap.String("service", serviceName), zap.String("from", oldPrimary), zap.String("to", newPrimary)) entries, err := discovery.GetMSMembers(serviceName, client) if err != nil { @@ -213,17 +284,25 @@ func TransferPrimary(client *clientv3.Client, p *member.Participant, serviceName return errors.Errorf("failed to grant lease for expected primary, err: %v", err) } + primaryID := primaryIDs[nextPrimaryID] msParam := &keypath.MsParam{ ServiceName: serviceName, GroupID: keyspaceGroupID, } primary := &primaryData{ - raw: primaryIDs[nextPrimaryID], - output: primaryIDs[nextPrimaryID], + raw: primaryID, + output: primaryID, } // Mark the expected primary first so the affinity guard is in place before the - // current primary releases the leader key below. - if err = markExpectedPrimaryFlag(client, msParam, primary, grantResp.ID); err != nil { + // current primary releases the leader key below. leaderKeyGuard was built from the + // CreateRevision captured right after the IsServing() check above, before discovery + // and this lease grant, so it fences the write to that exact leadership session for + // the whole function, not just the gap between here and the commit. It is evaluated + // atomically with the Put inside the same etcd transaction, so anything that + // changed the leader key since - a fresh campaign included, even by this same + // participant with the same MemberValue() - is caught at commit time. + if err = markExpectedPrimaryFlag(client, msParam, primary, grantResp.ID, leaderKeyGuard); err != nil { + revokeExpectedPrimaryLease(client, grantResp.ID) return errors.Errorf("failed to mark expected primary flag for %s, err: %v", serviceName, err) } @@ -231,9 +310,38 @@ func TransferPrimary(client *clientv3.Client, p *member.Participant, serviceName // IsServing() flip to false immediately, so the primary election loop steps down // and re-campaigns, where the affinity guard routes the leadership to the target. p.Resign() + + // Report success here without waiting for primaryID to actually win the campaign + // or finish initializing. This means the worst-case unavailability window this + // call can leave behind - if primaryID never wins a single campaign at all (down, + // unreachable, or stuck) - is bounded by expectedLease, i.e. exactly one leader + // lease (TransferPrimaryLeaseMultiplier == 1): that is how long the marker written + // above stays valid and keeps every other candidate skipping campaigns in favor of + // primaryID (see the affinity check next to GetExpectedPrimaryFlag in each + // service's election loop). Once the TTL lapses the marker disappears and any live + // member is free to win instead, so a completely absent target costs at most one + // lease of downtime, never longer. A target that does win - even if it + // subsequently fails to initialize and steps back down - clears the marker + // immediately on winning (see DeleteExpectedPrimaryFlag, called before + // initialization in campaignPrimary), so that path recovers well before the TTL + // and does not compound with this bound. return nil } +// revokeExpectedPrimaryLease revokes a lease granted for an expected-primary +// marker whose write failed or was rejected by the leadership guard, so a +// failed/aborted transfer never leaks a lease. Best effort: a failure here is +// logged, not propagated - the caller already has the original error to report, +// and the lease's own TTL bounds the residual leak if revoke also fails. +func revokeExpectedPrimaryLease(client *clientv3.Client, leaseID clientv3.LeaseID) { + ctx, cancel := context.WithTimeout(client.Ctx(), etcdutil.DefaultRequestTimeout) + defer cancel() + if _, err := client.Revoke(ctx, leaseID); err != nil { + log.Warn("failed to revoke expected primary lease after failed marker write", + zap.Int64("lease-id", int64(leaseID)), errs.ZapError(err)) + } +} + func isSamePrimary(member discovery.ServiceRegistryEntry, primary string) bool { return primary != "" && (member.Name == primary || member.ServiceAddr == primary) } diff --git a/pkg/mcs/utils/expected_primary_test.go b/pkg/mcs/utils/expected_primary_test.go index e10b84e5a2..92e55dc3c5 100644 --- a/pkg/mcs/utils/expected_primary_test.go +++ b/pkg/mcs/utils/expected_primary_test.go @@ -19,10 +19,14 @@ import ( "testing" "github.com/stretchr/testify/require" + clientv3 "go.etcd.io/etcd/client/v3" "go.uber.org/goleak" + "github.com/pingcap/kvproto/pkg/schedulingpb" + "github.com/tikv/pd/pkg/mcs/discovery" "github.com/tikv/pd/pkg/mcs/utils/constant" + "github.com/tikv/pd/pkg/member" "github.com/tikv/pd/pkg/utils/etcdutil" "github.com/tikv/pd/pkg/utils/keypath" "github.com/tikv/pd/pkg/utils/testutil" @@ -61,7 +65,7 @@ func TestDeleteExpectedPrimaryFlagRevokesLease(t *testing.T) { re.Equal(int64(leaseID), getResp.Kvs[0].Lease) // The newly elected primary cleans up the flag once it wins. - DeleteExpectedPrimaryFlag(client, msParam, value) + re.False(DeleteExpectedPrimaryFlag(client, msParam, value, nil)) // The key is gone... getResp, err = client.Get(ctx, path) @@ -76,8 +80,9 @@ func TestDeleteExpectedPrimaryFlagRevokesLease(t *testing.T) { // TestDeleteExpectedPrimaryFlagSkipsOnValueMismatch ensures the conditional delete // does not clobber a newer transfer. If a second transfer has already rewritten the -// flag (with its own lease) while this primary was winning, the stale winner must -// leave both the key and the newer lease intact. +// flag (with its own lease) to another member while this primary was winning, the +// stale winner must leave both the key and the newer lease intact and report that +// it has been superseded, so it steps down instead of defeating that transfer. func TestDeleteExpectedPrimaryFlagSkipsOnValueMismatch(t *testing.T) { re := require.New(t) _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) @@ -93,8 +98,9 @@ func TestDeleteExpectedPrimaryFlagSkipsOnValueMismatch(t *testing.T) { leaseID := grantResp.ID re.NoError(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "newer", output: "newer"}, leaseID)) - // A primary that campaigned for the older value tries to clean up; it must not. - DeleteExpectedPrimaryFlag(client, msParam, "older") + // A primary that campaigned for the older value tries to clean up; it must not + // delete the newer marker, and it must learn that it has been superseded. + re.True(DeleteExpectedPrimaryFlag(client, msParam, "older", nil)) getResp, err := client.Get(ctx, path) re.NoError(err) @@ -105,6 +111,250 @@ func TestDeleteExpectedPrimaryFlagSkipsOnValueMismatch(t *testing.T) { re.Positive(ttlResp.TTL) } +// TestDeleteExpectedPrimaryFlagReconciliation covers the post-campaign reconcile +// paths of DeleteExpectedPrimaryFlag beyond the plain delete/mismatch cases: +// - a winner of a free election (no marker observed) keeps serving when the +// marker is still absent, but steps down when a newer transfer installed a +// marker naming another member while it was winning; +// - when the newer marker targets the winner itself, the winner keeps serving +// and the marker is cleaned up together with its lease. +func TestDeleteExpectedPrimaryFlagReconciliation(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + + ctx := context.Background() + msParam := &keypath.MsParam{ServiceName: constant.SchedulingServiceName} + path := keypath.ExpectedPrimaryPath(msParam) + const selfURL = "http://127.0.0.1:2379" + + self := member.NewParticipant(client, *msParam) + self.InitInfo(&schedulingpb.Participant{ + Name: "self", + Id: 1, + ListenUrls: []string{selfURL}, + }, "primary election") + + // Free election, no marker: nothing to reconcile, keep serving. + re.False(DeleteExpectedPrimaryFlag(client, msParam, "", self)) + + // A transfer installs a marker naming another member after the free election + // was won: the winner must step down, leaving the marker and its lease intact. + grantResp, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + re.NoError(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "http://other:2379", output: "other"}, grantResp.ID)) + re.True(DeleteExpectedPrimaryFlag(client, msParam, "", self)) + getResp, err := client.Get(ctx, path) + re.NoError(err) + re.Len(getResp.Kvs, 1) + + // The marker is rewritten to target the winner itself: keep serving, and the + // marker plus its lease are cleaned up. + grantResp2, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + re.NoError(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: selfURL, output: "self"}, grantResp2.ID)) + re.False(DeleteExpectedPrimaryFlag(client, msParam, "", self)) + getResp, err = client.Get(ctx, path) + re.NoError(err) + re.Empty(getResp.Kvs) + ttlResp, err := client.TimeToLive(ctx, grantResp2.ID) + re.NoError(err) + re.Equal(int64(-1), ttlResp.TTL) +} + +// TestExpectedPrimaryCmp verifies the campaign guard returned by ExpectedPrimaryCmp. +// The empty-value case must assert the marker is still absent: this closes the race +// where a campaigner that observed no transfer resumes after a transfer installed a +// target marker and released the leader key, and would otherwise win the campaign +// while bypassing the transfer affinity guard. +func TestExpectedPrimaryCmp(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + + ctx := context.Background() + msParam := &keypath.MsParam{ServiceName: constant.SchedulingServiceName} + path := keypath.ExpectedPrimaryPath(msParam) + + // runGuarded runs a txn gated only by the campaign guard and reports whether it + // committed, mirroring how CampaignWithCmps folds the guard into the campaign txn. + runGuarded := func(expectedValue string) bool { + resp, err := client.Txn(ctx). + If(ExpectedPrimaryCmp(msParam, expectedValue)). + Then(clientv3.OpGet(path)). + Commit() + re.NoError(err) + return resp.Succeeded + } + + // No transfer in progress: the empty-value guard (marker still absent) holds. + re.True(runGuarded("")) + + // A transfer installs a marker naming "target". + grantResp, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + re.NoError(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "target", output: "target"}, grantResp.ID)) + + // The empty-value guard must now fail: a campaigner that observed "no transfer" + // can no longer win once a marker exists. + re.False(runGuarded("")) + // The transfer target's own guard (value match) still holds, so it can campaign. + re.True(runGuarded("target")) + // A stale or non-target value does not match the installed marker. + re.False(runGuarded("other")) +} + +// TestMarkExpectedPrimaryFlagGuardsAgainstLeadershipChange reconstructs the race +// flagged in review: TransferPrimary checks IsServing() before doing discovery and +// lease-grant work, but only writes the expected-primary marker afterwards. If this +// instance loses leadership in that window - because it resigned, its lease +// expired, or anything else vacated the leader key - a rival can win a fresh +// campaign and start serving before the stale caller's marker write lands. That +// marker write must not silently succeed once a different primary is already +// serving: the guarded Put must be rejected by the same leader-key comparison +// election.Leadership uses to guard its own writes, and the marker must never +// appear in etcd. +func TestMarkExpectedPrimaryFlagGuardsAgainstLeadershipChange(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + + ctx := context.Background() + msParam := &keypath.MsParam{ServiceName: constant.SchedulingServiceName} + path := keypath.ExpectedPrimaryPath(msParam) + const selfURL = "http://127.0.0.1:2379" + const rivalURL = "http://127.0.0.1:2380" + + self := member.NewParticipant(client, *msParam) + self.InitInfo(&schedulingpb.Participant{ + Name: "self", + Id: 1, + ListenUrls: []string{selfURL}, + }, "primary election") + + // self wins the campaign and starts serving, exactly like the real election + // loop: CampaignWithCmps writes the leader key, then PromoteSelf marks it locally. + re.NoError(self.CampaignWithCmps(ctx, constant.DefaultLease)) + self.PromoteSelf() + re.True(self.IsServing()) + + // Snapshot the guard TransferPrimary would build right after its IsServing() + // check, while self still legitimately owns the leader key. + guard := clientv3.Compare(clientv3.Value(self.GetLeadership().GetLeaderKey()), "=", self.MemberValue()) + + // self loses leadership in the window between the check and the marker write + // (resign releases the leader key and revokes its lease)... + self.Resign() + + // ...and a rival wins a fresh campaign on the same election, becoming the new + // primary that will never look at a marker addressed by the stale request below. + rival := member.NewParticipant(client, *msParam) + rival.InitInfo(&schedulingpb.Participant{ + Name: "rival", + Id: 2, + ListenUrls: []string{rivalURL}, + }, "primary election") + re.NoError(rival.CampaignWithCmps(ctx, constant.DefaultLease)) + rival.PromoteSelf() + re.True(rival.IsServing()) + + // The stale request proceeds as TransferPrimary does after its initial check: + // grant a lease and attempt the guarded marker write with the snapshotted guard. + grantResp, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + err = markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "http://third:2379", output: "third"}, grantResp.ID, guard) + re.Error(err) + + // The marker must never have been published: the new primary is already serving + // and would never react to it. + getResp, err := client.Get(ctx, path) + re.NoError(err) + re.Empty(getResp.Kvs) + + // Mirror TransferPrimary's own cleanup on this failure path so the granted lease + // doesn't leak past the test. + _, err = client.Revoke(ctx, grantResp.ID) + re.NoError(err) +} + +// TestMarkExpectedPrimaryFlagGuardFencesElectionTerm covers the gap flagged in +// review on top of TestMarkExpectedPrimaryFlagGuardsAgainstLeadershipChange: +// Participant.MemberValue() is fixed for the participant's lifetime, so a guard +// built from Value(leaderKey) == MemberValue() cannot tell "the same leadership +// term IsServing() observed" apart from "this member lost its lease and won a +// fresh campaign with the same MemberValue()" while a transfer request was stalled +// in discovery or the lease grant. TransferPrimary guards on the leader key's +// CreateRevision instead, which changes on every fresh campaign (Campaign requires +// CreateRevision(leaderKey) == 0 to win), so it fences to the exact etcd write +// backing the observed term. This test reconstructs that race directly against +// markExpectedPrimaryFlag: the same participant loses and regains leadership +// (same MemberValue, new term), and a guard captured before the regain must be +// rejected, while a guard captured after it must succeed. +func TestMarkExpectedPrimaryFlagGuardFencesElectionTerm(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + + ctx := context.Background() + msParam := &keypath.MsParam{ServiceName: constant.SchedulingServiceName} + path := keypath.ExpectedPrimaryPath(msParam) + const selfURL = "http://127.0.0.1:2379" + + self := member.NewParticipant(client, *msParam) + self.InitInfo(&schedulingpb.Participant{ + Name: "self", + Id: 1, + ListenUrls: []string{selfURL}, + }, "primary election") + leaderKeyPath := self.GetLeadership().GetLeaderKey() + + // self wins the first term and starts serving. + re.NoError(self.CampaignWithCmps(ctx, constant.DefaultLease)) + self.PromoteSelf() + re.True(self.IsServing()) + + // Snapshot the guard the way TransferPrimary does: read the leader key live and + // fence on its CreateRevision, right after the IsServing() check would have run. + staleResp, err := client.Get(ctx, leaderKeyPath) + re.NoError(err) + re.Len(staleResp.Kvs, 1) + re.Equal(self.MemberValue(), string(staleResp.Kvs[0].Value)) + staleGuard := clientv3.Compare(clientv3.CreateRevision(leaderKeyPath), "=", staleResp.Kvs[0].CreateRevision) + + // self loses leadership (lease expiry, an unrelated resign, etc.) and then wins a + // fresh campaign - a new term, same MemberValue since it never changes after + // InitInfo. This is exactly the window a value-only guard cannot see through. + self.Resign() + re.NoError(self.CampaignWithCmps(ctx, constant.DefaultLease)) + self.PromoteSelf() + re.True(self.IsServing()) + + freshResp, err := client.Get(ctx, leaderKeyPath) + re.NoError(err) + re.Len(freshResp.Kvs, 1) + re.Equal(self.MemberValue(), string(freshResp.Kvs[0].Value)) + re.NotEqual(staleResp.Kvs[0].CreateRevision, freshResp.Kvs[0].CreateRevision, "re-campaigning must produce a new CreateRevision") + + // The stale, pre-regain guard must be rejected even though the value matches. + grantResp, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + re.Error(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "http://third:2379", output: "third"}, grantResp.ID, staleGuard)) + getResp, err := client.Get(ctx, path) + re.NoError(err) + re.Empty(getResp.Kvs) + _, err = client.Revoke(ctx, grantResp.ID) + re.NoError(err) + + // A guard captured after the regain, against the current term, succeeds. + freshGuard := clientv3.Compare(clientv3.CreateRevision(leaderKeyPath), "=", freshResp.Kvs[0].CreateRevision) + grantResp2, err := client.Grant(ctx, constant.TransferPrimaryLeaseMultiplier*constant.DefaultLease) + re.NoError(err) + re.NoError(markExpectedPrimaryFlag(client, msParam, &primaryData{raw: "http://third:2379", output: "third"}, grantResp2.ID, freshGuard)) + getResp, err = client.Get(ctx, path) + re.NoError(err) + re.Len(getResp.Kvs, 1) +} + // TestIsSamePrimary covers the matching used by TransferPrimary to skip a // self-transfer (#10970): a member matches by either its name or its service // address, and an empty target never matches. diff --git a/pkg/tso/allocator.go b/pkg/tso/allocator.go index 6f92e7e0fc..ff4bf1d08a 100644 --- a/pkg/tso/allocator.go +++ b/pkg/tso/allocator.go @@ -279,10 +279,7 @@ func (a *Allocator) campaignPrimary(expectedPrimary string) { GroupID: a.keyspaceGroupID, } m := a.member.(*member.Participant) - var cmps []clientv3.Cmp - if cmp := mcsutils.ExpectedPrimaryCmp(msParam, expectedPrimary); cmp != nil { - cmps = append(cmps, *cmp) - } + cmps := []clientv3.Cmp{mcsutils.ExpectedPrimaryCmp(msParam, expectedPrimary)} if err := m.CampaignWithCmps(a.ctx, lease, cmps...); err != nil { if errors.Is(err, errs.ErrEtcdTxnConflict) { log.Info("campaign tso primary meets error due to txn conflict, another tso server may campaign successfully", @@ -313,8 +310,13 @@ func (a *Allocator) campaignPrimary(expectedPrimary string) { // We have won the campaign, so the expected primary flag (if any) has served its // purpose as the affinity guard. Delete it so steady state is clean and a later - // failure re-elects immediately instead of waiting for the flag's TTL. - mcsutils.DeleteExpectedPrimaryFlag(a.member.Client(), msParam, expectedPrimary) + // failure re-elects immediately instead of waiting for the flag's TTL. If a newer + // transfer rewrote the flag to another member while we were winning, step down so + // the re-election routes leadership to that target. + if mcsutils.DeleteExpectedPrimaryFlag(a.member.Client(), msParam, expectedPrimary, m) { + log.Info("the expected primary has been changed to another member, stepping down", a.logFields...) + return + } log.Info("initializing the tso allocator") if err := a.Initialize(); err != nil { diff --git a/tests/integrations/mcs/members/member_test.go b/tests/integrations/mcs/members/member_test.go index 1c2e1df411..689be1a437 100644 --- a/tests/integrations/mcs/members/member_test.go +++ b/tests/integrations/mcs/members/member_test.go @@ -632,6 +632,11 @@ func (suite *memberTestSuite) TestTransferPrimaryWhileLeaseExpired() { resp, err := tests.TestDialClient.Post(fmt.Sprintf("%s/%s/api/v1/primary/transfer", primary, strings.ReplaceAll(service, "_", "-")), "application/json", bytes.NewBuffer(data)) re.NoError(err) + // TransferPrimary reports success as soon as the marker is written and the + // old primary resigns - it does not wait for newPrimary to actually win, so + // this call succeeds here even though skipGrantLeader keeps newPrimary from + // winning for the whole duration of this call. The cluster does recover once + // skipGrantLeader is disabled below. re.Equal(http.StatusOK, resp.StatusCode) resp.Body.Close() @@ -693,6 +698,10 @@ func (suite *memberTestSuite) TestTransferPrimaryWhileLeaseExpiredAndServerDown( resp, err := tests.TestDialClient.Post(fmt.Sprintf("%s/%s/api/v1/primary/transfer", primary, strings.ReplaceAll(service, "_", "-")), "application/json", bytes.NewBuffer(data)) re.NoError(err) + // TransferPrimary reports success as soon as the marker is written and the + // old primary resigns, regardless of whether newPrimary ever actually wins - + // skipGrantLeader keeps it from winning for as long as this call runs, but + // that only affects whether the cluster later recovers, not this response. re.Equal(http.StatusOK, resp.StatusCode) resp.Body.Close() @@ -708,9 +717,9 @@ func (suite *memberTestSuite) TestTransferPrimaryWhileLeaseExpiredAndServerDown( re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/election/skipGrantLeader")) // The transfer target (which we just closed) is the one named in the expected - // primary flag, so the other replicas back off until the flag's TTL (a few - // leader leases) expires, after which a free re-election elects a live primary. - // Wait long enough to cover that worst case. + // primary flag, so the other replicas back off until the flag's TTL (one leader + // lease) expires, after which a free re-election elects a live primary. Wait + // long enough to cover that worst case. recoverWait := time.Duration(mcs.TransferPrimaryLeaseMultiplier*mcs.DefaultLease)*time.Second + 10*time.Second serving := make([]string, 0, len(nodes)) testutil.Eventually(re, func() bool { @@ -962,6 +971,10 @@ func (suite *memberTestSuite) TestTransferPrimaryWhileLeaseExpiredAndServerDownW fmt.Sprintf("%s/tso/api/v1/primary/transfer", groupPrimary), "application/json", bytes.NewBuffer(data)) re.NoError(err) + // TransferPrimary reports success as soon as the marker is written and the old + // primary resigns, regardless of whether target ever actually wins - target can + // never win while skipGrantLeader is enabled, but that only affects whether the + // group later recovers, verified separately below. re.Equal(http.StatusOK, resp.StatusCode) resp.Body.Close()