Fix data race in ReconnectClient (#185)

Author: at-wat

atomic.go

@@ -0,0 +1,38 @@
+// Copyright 2021 The mqtt-go 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 mqtt
+
+import (
+	"sync"
+)
+
+type firstError struct {
+	mu  sync.RWMutex
+	err error
+}
+
+func (e *firstError) Store(err error) {
+	e.mu.Lock()
+	if e.err == nil {
+		e.err = err
+	}
+	e.mu.Unlock()
+}
+
+func (e *firstError) Load() error {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+	return e.err
+}

atomic_test.go

@@ -0,0 +1,43 @@
+// Copyright 2021 The mqtt-go 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 mqtt
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestFirstError(t *testing.T) {
+	var fe firstError
+
+	if err := fe.Load(); err != nil {
+		t.Fatalf("Initial value must be 'nil', got '%v'", err)
+	}
+
+	errDummy0 := errors.New("dummy1")
+	errDummy1 := errors.New("dummy1")
+
+	fe.Store(errDummy0)
+
+	if err := fe.Load(); err != errDummy0 {
+		t.Fatalf("Expected '%v', got '%v'", errDummy0, err)
+	}
+
+	fe.Store(errDummy1)
+
+	if err := fe.Load(); err != errDummy0 {
+		t.Fatalf("Value is updated after first store. Expected '%v', got '%v'", errDummy0, err)
+	}
+}

reconnclient.go

@@ -69,7 +69,7 @@ func (c *reconnectClient) Connect(ctx context.Context, clientID string, opts ...
 		c.options.Timeout = c.options.PingInterval
 	}
 
-	var errDial, errConnect error
+	var errDial, errConnect firstError
 
 	done := make(chan struct{})
 	var doneOnce sync.Once
@@ -136,12 +136,12 @@ func (c *reconnectClient) Connect(ctx context.Context, clientID string, opts ...
 					case <-c.disconnected:
 						return
 					}
-				} else if err != ctxTimeout.Err() || errConnect == nil {
-					errConnect = err // Hold last connect error but avoid overwrote by context cancel.
+				} else if err != ctxTimeout.Err() {
+					errConnect.Store(err) // Hold first connect error excepting context cancel.
 				}
 				cancel()
-			} else if err != ctx.Err() || errDial == nil {
-				errDial = err // Hold last dial error but avoid overwrote by context cancel.
+			} else if err != ctx.Err() {
+				errDial.Store(err) // Hold first dial error excepting context cancel.
 			}
 			select {
 			case <-time.After(reconnWait):
@@ -161,11 +161,11 @@ func (c *reconnectClient) Connect(ctx context.Context, clientID string, opts ...
 	case <-done:
 	case <-ctx.Done():
 		var actualErrs []string
-		if errDial != nil {
-			actualErrs = append(actualErrs, fmt.Sprintf("dial: %v", errDial))
+		if err := errDial.Load(); err != nil {
+			actualErrs = append(actualErrs, fmt.Sprintf("dial: %v", err))
 		}
-		if errConnect != nil {
-			actualErrs = append(actualErrs, fmt.Sprintf("connect: %v", errConnect))
+		if err := errConnect.Load(); err != nil {
+			actualErrs = append(actualErrs, fmt.Sprintf("connect: %v", err))
 		}
 		var errStr string
 		if len(actualErrs) > 0 {