fix(remoting): use 3-state atomic + sync.Cond for ExchangeClient.init

Address Copilot review feedback:
- Replace uatomic.Bool with uatomic.Int32 for 3-state init tracking
  (0=uninitialized, 1=initializing, 2=initialized) to prevent callers
  from observing a partially-initialized state
- Replace spin-wait with sync.Cond for proper coordination: waiters
  block until signaled instead of polling
- All concurrent waiters now receive the same error result when init
  fails, rather than silently returning nil
- Close() properly resets state and broadcasts to wake any waiters
- Fix test: use buffered channel for concurrent error collection instead
  of calling t.Errorf from multiple goroutines

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Aias00

remoting/exchange_client.go

@@ -19,6 +19,7 @@ package remoting
 
 import (
 	"errors"
+	"sync"
 	"time"
 )
 
@@ -61,7 +62,9 @@ type ExchangeClient struct {
 	ConnectTimeout time.Duration  // timeout for connecting server
 	address        string         // server address for dialing. The format: ip:port
 	client         Client         // dealing with the transport
-	init           uatomic.Bool  // the tag for init, protected by atomic operations
+	initState      uatomic.Int32  // init state: 0=uninitialized, 1=initializing, 2=initialized
+	initCond       *sync.Cond     // signals waiters when init completes (success or failure)
+	initErr        error          // stored init error for waiters to observe (guarded by initCond.L)
 	activeNum      uatomic.Uint32 // the number of service using the exchangeClient
 }
 
@@ -71,6 +74,7 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 		ConnectTimeout: connectTimeout,
 		address:        url.Location,
 		client:         client,
+		initCond:       sync.NewCond(&sync.Mutex{}),
 	}
 	if !lazyInit {
 		if err := exchangeClient.doInit(url); err != nil {
@@ -82,29 +86,43 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 }
 
 func (cl *ExchangeClient) doInit(url *common.URL) error {
-	if cl.init.Load() {
+	// Fast path: already initialized.
+	if cl.initState.Load() == 2 {
 		return nil
 	}
-	// Use CompareAndSwap to ensure only one goroutine performs initialization.
-	// If CAS fails, another goroutine already initialized; spin-wait until init completes.
-	if !cl.init.CAS(false, true) {
-		// Another goroutine is initializing; wait for it to complete.
-		for !cl.init.Load() {
-			time.Sleep(10 * time.Millisecond)
+	// Try to become the initializer: transition 0 -> 1 (uninitialized -> initializing).
+	if !cl.initState.CAS(0, 1) {
+		// Another goroutine is initializing. Wait for it to complete.
+		cl.initCond.L.Lock()
+		for cl.initState.Load() == 1 {
+			cl.initCond.Wait()
 		}
-		return nil
+		err := cl.initErr
+		cl.initCond.L.Unlock()
+		return err
 	}
 	// This goroutine won the CAS — perform the actual initialization.
+	var initErr error
 	if cl.client.Connect(url) != nil {
 		// retry for a while
 		time.Sleep(100 * time.Millisecond)
 		if cl.client.Connect(url) != nil {
 			logger.Errorf("[Remoting] failed to connect server, url=%v", url.Location)
-			cl.init.Store(false) // reset on failure so future calls can retry
-			return errors.New("Failed to connect server " + url.Location)
+			initErr = errors.New("Failed to connect server " + url.Location)
 		}
 	}
-	return nil
+	cl.initCond.L.Lock()
+	if initErr != nil {
+		// Reset to uninitialized so future calls can retry.
+		cl.initState.Store(0)
+		cl.initErr = initErr
+	} else {
+		cl.initState.Store(2)
+		cl.initErr = nil
+	}
+	cl.initCond.L.Unlock()
+	cl.initCond.Broadcast()
+	return initErr
 }
 
 // IncreaseActiveNumber increase number of service using client.
@@ -204,7 +222,11 @@ func (client *ExchangeClient) Send(invocation *base.Invocation, url *common.URL,
 // Close close the client.
 func (client *ExchangeClient) Close() {
 	client.client.Close()
-	client.init.Store(false)
+	client.initCond.L.Lock()
+	client.initState.Store(0)
+	client.initErr = nil
+	client.initCond.L.Unlock()
+	client.initCond.Broadcast()
 }
 
 // IsAvailable to check if the underlying network client is available yet.

remoting/exchange_client_test.go

@@ -92,7 +92,7 @@ func TestExchangeClientClose(t *testing.T) {
 	m := &mockClient{available: true}
 	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
 	ec.Close()
-	assert.False(t, ec.init.Load())
+	assert.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after Close")
 }
 
 func TestExchangeClientIsAvailable(t *testing.T) {
@@ -112,39 +112,72 @@ func TestExchangeClientConcurrentDoInit(t *testing.T) {
 
 	var wg sync.WaitGroup
 	const goroutines = 20
+	errCh := make(chan error, goroutines)
 	wg.Add(goroutines)
 	for i := 0; i < goroutines; i++ {
 		go func() {
 			defer wg.Done()
-			// All goroutines trigger lazy init concurrently via Request-like paths
 			if err := ec.doInit(testURL()); err != nil {
-				t.Errorf("doInit failed: %v", err)
+				errCh <- err
 			}
 		}()
 	}
 	wg.Wait()
+	close(errCh)
+
+	for err := range errCh {
+		t.Errorf("doInit failed: %v", err)
+	}
 
 	m.mu.Lock()
 	count := m.connCount
 	m.mu.Unlock()
 	assert.Equal(t, 1, count, "Connect should be called exactly once despite concurrent doInit")
-	assert.True(t, ec.init.Load(), "init flag should be true after doInit")
+	assert.Equal(t, int32(2), ec.initState.Load(), "initState should be 2 (initialized) after doInit")
 }
 
 func TestExchangeClientDoInitFailureResets(t *testing.T) {
-	// Verify that a failed doInit resets the init flag so future calls can retry.
+	// Verify that a failed doInit resets the init state so future calls can retry.
 	m := &mockClient{connectErr: errors.New("fail")}
 	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
 
 	err := ec.doInit(testURL())
 	assert.Error(t, err)
-	assert.False(t, ec.init.Load(), "init flag should be reset after failed doInit")
+	assert.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after failed doInit")
 
 	// Now make Connect succeed and retry
 	m.mu.Lock()
 	m.connectErr = nil
 	m.mu.Unlock()
 	err = ec.doInit(testURL())
 	assert.NoError(t, err)
-	assert.True(t, ec.init.Load(), "init flag should be true after successful retry")
+	assert.Equal(t, int32(2), ec.initState.Load(), "initState should be 2 (initialized) after successful retry")
+}
+
+func TestExchangeClientConcurrentDoInitPropagatesError(t *testing.T) {
+	// Verify that when initialization fails, all concurrent waiters receive the same error.
+	m := &mockClient{connectErr: errors.New("connect refused")}
+	ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+	var wg sync.WaitGroup
+	const goroutines = 10
+	errCh := make(chan error, goroutines)
+	wg.Add(goroutines)
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			errCh <- ec.doInit(testURL())
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+
+	errorCount := 0
+	for err := range errCh {
+		if err != nil {
+			errorCount++
+		}
+	}
+	assert.Equal(t, goroutines, errorCount, "all goroutines should receive the init error")
+	assert.Equal(t, int32(0), ec.initState.Load(), "initState should be 0 (uninitialized) after failure")
 }