Improve reconnect and resubscribe stability (#108)

Several problems were found when testing library under unstable conditions – with frequent reconnections. Allows to avoid already subscribed subscription error, and bad request error (sending subscribe frame before connect frame). Also protects from running several simultaneous timers for reconnect/resubscribe.

Author: FZambia

client.go

@@ -601,6 +601,13 @@ func (c *Client) moveToConnecting(code uint32, reason string) {
 
 func (c *Client) scheduleReconnectLocked() {
 	c.reconnectAttempts++
+	if c.reconnectTimer != nil {
+		if c.logLevelEnabled(LogLevelDebug) {
+			c.log(LogLevelDebug, "stopping previous reconnect timer", nil)
+		}
+		c.reconnectTimer.Stop()
+		c.reconnectTimer = nil
+	}
 	reconnectDelay := c.getReconnectDelay()
 	if c.logLevelEnabled(LogLevelDebug) {
 		c.log(LogLevelDebug, "reconnect with delay", map[string]string{

examples/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.0 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
 	github.com/segmentio/encoding v0.4.0 // indirect
+	github.com/shadowspore/fossil-delta v0.0.0-20241213113458-1d797d70cbe3 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	golang.org/x/sys v0.11.0 // indirect
 	google.golang.org/protobuf v1.36.6 // indirect

examples/go.sum

@@ -22,8 +22,10 @@ github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
 github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
 github.com/segmentio/encoding v0.4.0 h1:MEBYvRqiUB2nfR2criEXWqwdY6HJOUrCn5hboVOVmy8=
 github.com/segmentio/encoding v0.4.0/go.mod h1:/d03Cd8PoaDeceuhUUUQWjU0KhWjrmYrWPgtJHYZSnI=
-github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
-github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/shadowspore/fossil-delta v0.0.0-20241213113458-1d797d70cbe3 h1:/4/IJi5iyTdh6mqOUaASW148HQpujYiHl0Wl78dSOSc=
+github.com/shadowspore/fossil-delta v0.0.0-20241213113458-1d797d70cbe3/go.mod h1:aJIMhRsunltJR926EB2MUg8qHemFQDreSB33pyto2Ps=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=

subscription.go

@@ -110,6 +110,8 @@ type Subscription struct {
 	deltaType       DeltaType
 	deltaNegotiated bool
 	prevData        []byte
+
+	inflight atomic.Bool
 }
 
 func (s *Subscription) State() SubState {
@@ -572,6 +574,15 @@ func (s *Subscription) applyDeltaLocked(pub *protocol.Publication, event Publica
 
 // Lock must be held outside.
 func (s *Subscription) scheduleResubscribe() {
+	if s.resubscribeTimer != nil {
+		if s.centrifuge.logLevelEnabled(LogLevelDebug) {
+			s.centrifuge.log(LogLevelDebug, "stopping previous resubscribe timer", map[string]string{
+				"channel": s.Channel,
+			})
+		}
+		s.resubscribeTimer.Stop()
+		s.resubscribeTimer = nil
+	}
 	delay := s.resubscribeStrategy.timeBeforeNextAttempt(s.resubscribeAttempts)
 	s.resubscribeAttempts++
 	s.resubscribeTimer = time.AfterFunc(delay, func() {
@@ -698,23 +709,36 @@ func (s *Subscription) resubscribe() {
 		s.mu.Unlock()
 		return
 	}
+	if s.inflight.Load() {
+		s.mu.Unlock()
+		if s.centrifuge.logLevelEnabled(LogLevelDebug) {
+			s.centrifuge.log(LogLevelDebug, "avoid subscribe since inflight", map[string]string{
+				"channel": s.Channel,
+			})
+		}
+		return
+	}
 	token := s.token
+	s.inflight.Store(true)
 	s.mu.Unlock()
 
 	if token == "" && s.getToken != nil {
 		var err error
 		token, err = s.getSubscriptionToken(s.Channel)
 		if err != nil {
 			if errors.Is(err, ErrUnauthorized) {
+				s.inflight.Store(false)
 				s.unsubscribe(unsubscribedUnauthorized, "unauthorized", false)
 				return
 			}
+			s.inflight.Store(false)
 			s.subscribeError(err)
 			return
 		}
 		s.mu.Lock()
 		if token == "" {
 			s.mu.Unlock()
+			s.inflight.Store(false)
 			s.unsubscribe(unsubscribedUnauthorized, "unauthorized", false)
 			return
 		}
@@ -725,6 +749,7 @@ func (s *Subscription) resubscribe() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.state != SubStateSubscribing {
+		s.inflight.Store(false)
 		return
 	}
 
@@ -738,12 +763,15 @@ func (s *Subscription) resubscribe() {
 
 	err := s.centrifuge.sendSubscribe(s.Channel, s.data, isRecover, sp, token, s.positioned, s.recoverable, s.joinLeave, s.deltaType, func(res *protocol.SubscribeResult, err error) {
 		if err != nil {
+			s.inflight.Store(false)
 			s.subscribeError(err)
 			return
 		}
+		s.inflight.Store(false)
 		s.moveToSubscribed(res)
 	})
 	if err != nil {
+		s.inflight.Store(false)
 		s.scheduleResubscribe()
 	}
 }