fix concurrent issue

Author: noel2004

coordinator/internal/controller/proxy/auth.go

@@ -1,7 +1,10 @@
 package proxy
 
 import (
+	"context"
 	"fmt"
+	"sync"
+	"time"
 
 	jwt "github.com/appleboy/gin-jwt/v2"
 	"github.com/gin-gonic/gin"
@@ -21,6 +24,7 @@ type AuthController struct {
 	proverMgr *ProverManager
 }
 
+const upstreamConnTimeout = time.Second * 5
 const LoginParamCache = "login_param"
 const ProverTypesKey = "prover_types"
 const SignatureKey = "prover_signature"
@@ -59,17 +63,25 @@ func (a *AuthController) Login(c *gin.Context) (interface{}, error) {
 	session := a.proverMgr.GetOrCreate(loginParam.PublicKey)
 	log.Debug("start handling login", "cli", loginParam.Message.ProverName)
 
+	loginCtx, cf := context.WithTimeout(context.Background(), upstreamConnTimeout)
+	var wg sync.WaitGroup
 	for _, cli := range a.clients {
-
+		wg.Add(1)
 		go func(cli Client) {
-			if err := session.ProxyLogin(c, cli, &loginParam.LoginParameter); err != nil {
+			defer wg.Done()
+			if err := session.ProxyLogin(loginCtx, cli, &loginParam.LoginParameter); err != nil {
 				log.Error("proxy login failed during token cache update",
 					"userKey", loginParam.PublicKey,
 					"upstream", cli.Name(),
 					"error", err)
 			}
 		}(cli)
 	}
+	go func(cliName string) {
+		wg.Wait()
+		cf()
+		log.Debug("first login attempt has completed", "cli", cliName)
+	}(loginParam.Message.ProverName)
 
 	return loginParam.LoginParameter, nil
 }

coordinator/internal/controller/proxy/client_manager.go

@@ -123,10 +123,10 @@ func (cliMgr *ClientManager) doLogin(ctx context.Context, loginCli *upClient) {
 func (cliMgr *ClientManager) Reset(cli *upClient) {
 	cliMgr.cachedCli.Lock()
 	if cliMgr.cachedCli.cli == cli {
+		log.Info("cached client cleared", "name", cliMgr.name)
 		cliMgr.cachedCli.cli = nil
 	}
 	cliMgr.cachedCli.Unlock()
-	log.Info("cached client cleared", "name", cliMgr.name)
 }
 
 func (cliMgr *ClientManager) Name() string {

coordinator/internal/controller/proxy/prover_session.go

@@ -162,6 +162,7 @@ func (c *proverSession) maintainLogin(ctx context.Context, cliMgr Client, up str
 	}
 
 	if resp.ErrCode == ctypes.ErrJWTTokenExpired {
+		log.Info("up stream has expired, renew upstream connection", "up", up)
 		cliMgr.Reset(cli)
 		cli = cliMgr.Client(ctx)
 		if cli == nil {