@@ -66,16 +66,27 @@ type keyspaceGroupSvcDiscovery struct {
6666 secondaryURLs []string
6767 // urls are the primary/secondary serving URL
6868 urls []string
69+
70+ // version is used to avoid updating stale info
71+ version uint64
72+ }
73+
74+ func (k * keyspaceGroupSvcDiscovery ) getVersion () uint64 {
75+ return k .version
6976}
7077
7178func (k * keyspaceGroupSvcDiscovery ) update (
7279 keyspaceGroup * tsopb.KeyspaceGroup ,
7380 newPrimaryURL string ,
7481 secondaryURLs , urls []string ,
75- ) (oldPrimaryURL string , primarySwitched , secondaryChanged bool ) {
82+ version uint64 ,
83+ ) (oldPrimaryURL string , primarySwitched , success bool ) {
7684 k .Lock ()
7785 defer k .Unlock ()
78-
86+ if k .version >= version {
87+ return "" , false , false
88+ }
89+ success = true
7990 // If the new primary URL is empty, we don't switch the primary URL.
8091 oldPrimaryURL = k .primaryURL
8192 if len (newPrimaryURL ) > 0 {
@@ -85,11 +96,11 @@ func (k *keyspaceGroupSvcDiscovery) update(
8596
8697 if ! reflect .DeepEqual (k .secondaryURLs , secondaryURLs ) {
8798 k .secondaryURLs = secondaryURLs
88- secondaryChanged = true
8999 }
90100
91101 k .group = keyspaceGroup
92102 k .urls = urls
103+ k .version = version
93104 return
94105}
95106
@@ -160,6 +171,7 @@ func NewTSOServiceDiscovery(
160171 primaryURL : "" ,
161172 secondaryURLs : make ([]string , 0 ),
162173 urls : make ([]string , 0 ),
174+ version : 0 ,
163175 }
164176 c .tsoServerDiscovery = & tsoServerDiscovery {urls : make ([]string , 0 )}
165177 // Start with the default keyspace group. The actual keyspace group, to which the keyspace belongs,
@@ -413,12 +425,16 @@ func (c *tsoServiceDiscovery) updateMember() error {
413425
414426 keyspaceID := c .GetKeyspaceID ()
415427 var keyspaceGroup * tsopb.KeyspaceGroup
428+ var version uint64
429+ curVersion := c .keyspaceGroupSD .getVersion ()
416430 if len (tsoServerURL ) > 0 {
417- keyspaceGroup , err = c .findGroupByKeyspaceID (keyspaceID , tsoServerURL , UpdateMemberTimeout )
431+ keyspaceGroup , version , err = c .findGroupByKeyspaceID (keyspaceID , tsoServerURL , UpdateMemberTimeout , c . keyspaceGroupSD . getVersion () )
418432 if err != nil {
419433 log .Error ("[tso] failed to find the keyspace group" ,
420434 zap .Uint32 ("keyspace-id-in-request" , keyspaceID ),
421435 zap .String ("tso-server-url" , tsoServerURL ),
436+ zap .Uint64 ("response-version" , version ),
437+ zap .Uint64 ("current-version" , curVersion ),
422438 errs .ZapError (err ))
423439 return err
424440 }
@@ -451,6 +467,7 @@ func (c *tsoServiceDiscovery) updateMember() error {
451467 Id : constants .DefaultKeyspaceGroupID ,
452468 Members : members ,
453469 }
470+ version = curVersion + 1
454471 }
455472
456473 oldGroupID := c .GetKeyspaceGroupID ()
@@ -488,8 +505,12 @@ func (c *tsoServiceDiscovery) updateMember() error {
488505 }
489506 }
490507
491- oldPrimary , primarySwitched , _ :=
492- c .keyspaceGroupSD .update (keyspaceGroup , primaryURL , secondaryURLs , urls )
508+ oldPrimary , primarySwitched , success :=
509+ c .keyspaceGroupSD .update (keyspaceGroup , primaryURL , secondaryURLs , urls , version )
510+ if ! success {
511+ log .Warn ("[tso] failed to update keyspace group, met stale keyspace group info" , zap .Uint64 ("version" , version ))
512+ return errors .Errorf ("keyspace group version is stale, current version: %d, new version: %d" , c .keyspaceGroupSD .getVersion (), version )
513+ }
493514 if primarySwitched {
494515 log .Info ("[tso] updated keyspace group service discovery info" ,
495516 zap .Uint32 ("keyspace-id-in-request" , keyspaceID ),
@@ -505,15 +526,14 @@ func (c *tsoServiceDiscovery) updateMember() error {
505526 if len (primaryURL ) == 0 {
506527 return errors .New ("no primary URL found" )
507528 }
508-
509529 return nil
510530}
511531
512532// Query the keyspace group info from the tso server by the keyspace ID. The server side will return
513533// the info of the keyspace group to which this keyspace belongs.
514534func (c * tsoServiceDiscovery ) findGroupByKeyspaceID (
515- keyspaceID uint32 , tsoSrvURL string , timeout time.Duration ,
516- ) (* tsopb.KeyspaceGroup , error ) {
535+ keyspaceID uint32 , tsoSrvURL string , timeout time.Duration , version uint64 ,
536+ ) (* tsopb.KeyspaceGroup , uint64 , error ) {
517537 failpoint .Inject ("unexpectedCallOfFindGroupByKeyspaceID" , func (val failpoint.Value ) {
518538 keyspaceToCheck , ok := val .(int )
519539 if ok && keyspaceID == uint32 (keyspaceToCheck ) {
@@ -525,7 +545,7 @@ func (c *tsoServiceDiscovery) findGroupByKeyspaceID(
525545
526546 cc , err := c .GetOrCreateGRPCConn (tsoSrvURL )
527547 if err != nil {
528- return nil , err
548+ return nil , 0 , err
529549 }
530550
531551 resp , err := tsopb .NewTSOClient (cc ).FindGroupByKeyspaceID (
@@ -536,24 +556,30 @@ func (c *tsoServiceDiscovery) findGroupByKeyspaceID(
536556 KeyspaceGroupId : constants .DefaultKeyspaceGroupID ,
537557 },
538558 KeyspaceId : keyspaceID ,
559+ Version : version ,
539560 })
540561 if err != nil {
541562 attachErr := errors .Errorf ("error:%s target:%s status:%s" ,
542563 err , cc .Target (), cc .GetState ().String ())
543- return nil , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
564+ return nil , 0 , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
544565 }
545566 if resp .GetHeader ().GetError () != nil {
546567 attachErr := errors .Errorf ("error:%s target:%s status:%s" ,
547568 resp .GetHeader ().GetError ().String (), cc .Target (), cc .GetState ().String ())
548- return nil , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
569+ return nil , 0 , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
549570 }
550571 if resp .KeyspaceGroup == nil {
551572 attachErr := errors .Errorf ("error:%s target:%s status:%s" ,
552573 "no keyspace group found" , cc .Target (), cc .GetState ().String ())
553- return nil , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
574+ return nil , 0 , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
575+ }
576+ if resp .Version < version {
577+ attachErr := errors .Errorf ("error:%s target:%s response version:%d current version:%d" ,
578+ "response version less than the given version" , cc .Target (), resp .Version , version )
579+ return nil , 0 , errs .ErrClientFindGroupByKeyspaceID .Wrap (attachErr ).GenWithStackByCause ()
554580 }
555581
556- return resp .KeyspaceGroup , nil
582+ return resp .KeyspaceGroup , resp . GetVersion (), nil
557583}
558584
559585func (c * tsoServiceDiscovery ) getTSOServer (sd ServiceDiscovery ) (string , error ) {
0 commit comments