Make locking issue in kick a bit more debuggable

Author: xaionaro

go.mod

@@ -321,7 +321,7 @@ require (
 	github.com/xaionaro-go/xfyne v0.0.0-20250615190411-4c96281f6e25
 	github.com/xaionaro-go/xlogrus v0.0.0-20250111150201-60557109545a
 	github.com/xaionaro-go/xpath v0.0.0-20250111145115-55f5728f643f
-	github.com/xaionaro-go/xsync v0.0.0-20250713122735-6e002588c3d0
+	github.com/xaionaro-go/xsync v0.0.0-20250816221754-5ff810dd0a2c
 	github.com/yutopp/go-flv v0.3.1
 	golang.org/x/crypto v0.39.0
 	google.golang.org/grpc v1.73.0

go.sum

@@ -1159,6 +1159,8 @@ github.com/xaionaro-go/xpath v0.0.0-20250111145115-55f5728f643f h1:ofxY1akRlVdJ/
 github.com/xaionaro-go/xpath v0.0.0-20250111145115-55f5728f643f/go.mod h1:f0DVcqddOy1RALOgXJHwpQnkp1u1yeBX/+A2+Bf4EGc=
 github.com/xaionaro-go/xsync v0.0.0-20250713122735-6e002588c3d0 h1:JDl35mgD3RsmEx7xbBCSONLceBCTFUWQvPUG5YsWrgU=
 github.com/xaionaro-go/xsync v0.0.0-20250713122735-6e002588c3d0/go.mod h1:FCpywNAl4a4hgzE8j7Z+TpdhBQi5WHxnI35jOrFpoQw=
+github.com/xaionaro-go/xsync v0.0.0-20250816221754-5ff810dd0a2c h1:d1jyWhrTnHhwk+16Y7fexlXQCqfvm0NCsDvdsg2mcCA=
+github.com/xaionaro-go/xsync v0.0.0-20250816221754-5ff810dd0a2c/go.mod h1:FCpywNAl4a4hgzE8j7Z+TpdhBQi5WHxnI35jOrFpoQw=
 github.com/xaionaro-go/youtube-live-chat-downloader/v2 v2.0.0-20250726224735-811554015ae8 h1:iRcIRlIryjDkgFoKJZQQWra/zv783CnKtXnpq9z7VTQ=
 github.com/xaionaro-go/youtube-live-chat-downloader/v2 v2.0.0-20250726224735-811554015ae8/go.mod h1:TrUogg8mrebgMD/JU094CmSXn3yKrt+CZjiDL3YtmMw=
 github.com/xaionaro-go/zerolog2belt v0.0.0-20241103164018-a3bc1ea487e5 h1:jAy7VLg8y8XE1R8jBte4PRDJzOaAE+sUfmttfB9ZcAY=

pkg/streamcontrol/kick/kick.go

@@ -160,50 +160,51 @@ func (k *Kick) keepAliveLoop(
 func (k *Kick) initChatHandler(
 	ctx context.Context,
 ) error {
-	if !k.ChatHandlerLocker.Lock(ctx) {
-		return ctx.Err()
-	}
+	return xsync.DoA1R1(ctx, &k.ChatHandlerLocker, k.initChatHandlerNoLock, ctx)
+}
 
+func (k *Kick) initChatHandlerNoLock(
+	ctx context.Context,
+) error {
 	chatHandler, err := k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
 	if err == nil {
-		k.ChatHandlerLocker.Unlock()
 		k.ChatHandler = chatHandler
 		return nil
 	}
 
-	go func() {
-		defer k.ChatHandlerLocker.Unlock()
-		defer logger.Debugf(ctx, "/initChatHandler")
-		for {
-			logger.Errorf(ctx, "unable to initialize chat handler: %v", err)
-			time.Sleep(time.Second)
-			select {
-			case <-k.CloseCtx.Done():
-				logger.Debugf(ctx, "initChatHandler: cancelled (case #1)")
-				return
-			case <-ctx.Done():
-				logger.Debugf(ctx, "initChatHandler: cancelled (case #2)")
-				return
-			default:
-			}
-			chatHandler, err = k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
-			if err == nil {
-				break
-			}
+	for {
+		logger.Errorf(ctx, "unable to initialize chat handler: %v", err)
+		time.Sleep(time.Second)
+		select {
+		case <-k.CloseCtx.Done():
+			logger.Debugf(ctx, "initChatHandler: cancelled (case #1)")
+			return fmt.Errorf("k.CloseCtx is closed: %w", k.CloseCtx.Err())
+		case <-ctx.Done():
+			logger.Debugf(ctx, "initChatHandler: cancelled (case #2)")
+			return fmt.Errorf("ctx is closed: %w", ctx.Err())
+		default:
+		}
+		chatHandler, err = k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
+		if err != nil {
+			logger.Debugf(ctx, "initChatHandler: unable to create a new chat handler: %v", err)
+			continue
 		}
 		k.ChatHandler = chatHandler
-	}()
-	return nil
+		return nil
+	}
 }
 
 func (k *Kick) onChatHandlerClose(
 	ctx context.Context,
 	h *ChatHandlerOBSOLETE,
 ) {
-	if !k.ChatHandlerLocker.Lock(ctx) {
-		return
-	}
-	defer k.ChatHandlerLocker.Unlock()
+	xsync.DoA2(ctx, &k.ChatHandlerLocker, k.onChatHandlerCloseNoLock, ctx, h)
+}
+
+func (k *Kick) onChatHandlerCloseNoLock(
+	ctx context.Context,
+	h *ChatHandlerOBSOLETE,
+) {
 	if h != k.ChatHandler {
 		logger.Errorf(ctx, "chat handler was already replaced")
 		return
@@ -550,11 +551,9 @@ func (k *Kick) GetAllCategories(
 func (k *Kick) tryGetChatHandler(
 	ctx context.Context,
 ) *ChatHandlerOBSOLETE {
-	if !k.ChatHandlerLocker.Lock(ctx) {
-		return nil
-	}
-	defer k.ChatHandlerLocker.Unlock()
-	return k.ChatHandler
+	return xsync.DoR1(ctx, &k.ChatHandlerLocker, func() *ChatHandlerOBSOLETE {
+		return k.ChatHandler
+	})
 }
 
 func (k *Kick) getChatHandler(
@@ -761,10 +760,11 @@ func (k *Kick) prepareNoLock(ctx context.Context) error {
 			err = fmt.Errorf("initChannelInfo: %w", err)
 			return
 		}
-		if err = k.initChatHandler(ctx); err != nil {
-			err = fmt.Errorf("initChatHandler: %w", err)
-			return
-		}
+		observability.Go(ctx, func(ctx context.Context) {
+			if err = k.initChatHandler(ctx); err != nil {
+				logger.Errorf(ctx, "initChatHandler: %v", err)
+			}
+		})
 	})
 	return err
 }
@@ -846,7 +846,7 @@ func (k *Kick) refreshAccessToken(
 
 	err = k.setToken(ctx, resp, time.Now())
 	if err != nil {
-		return fmt.Errorf("unable to set access token: %w")
+		return fmt.Errorf("unable to set access token: %w", err)
 	}
 
 	return nil