diff --git a/backend/internal/worker/worker_test.go b/backend/internal/worker/worker_test.go index aae550df..d9deb678 100644 --- a/backend/internal/worker/worker_test.go +++ b/backend/internal/worker/worker_test.go @@ -387,9 +387,9 @@ func TestRedisSweepPicksUpDueDelayedJob(t *testing.T) { db := newWorkerTestDB(t) mr, rdb := newMiniRedis(t) baseTime := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) - muNow := baseTime + var elapsed atomic.Int64 wp := newRedisWorkerPool(t, db, rdb, - WithNow(func() time.Time { return muNow }), + WithNow(func() time.Time { return baseTime.Add(time.Duration(elapsed.Load())) }), WithSweepInterval(50*time.Millisecond), WithBlockTimeout(50*time.Millisecond), ) @@ -420,7 +420,7 @@ func TestRedisSweepPicksUpDueDelayedJob(t *testing.T) { } // Advance the mock clock past the scheduled_at so the job becomes due. - muNow = baseTime.Add(1 * time.Second) + elapsed.Store(int64(time.Second)) // The sweep goroutine should now see the job as due, push it to Redis, // and the worker should process it. @@ -578,13 +578,14 @@ func TestRedisEnqueuePushFailureCompensatedBySweep(t *testing.T) { rdb2 := redis.NewClient(&redis.Options{Addr: mr2.Addr()}) t.Cleanup(func() { rdb2.Close() }) - // Swap in the recovered Redis client. - wp.mu.Lock() + // Restart with the recovered client so no worker can use it while it is replaced. + if err := wp.Stop(); err != nil { + t.Fatalf("stop before Redis replacement: %v", err) + } wp.rdb = rdb2 - wp.mu.Unlock() - - // Recreate consumer groups on the new Redis instance. - wp.ensureConsumerGroups(context.Background()) + if err := wp.Start(); err != nil { + t.Fatalf("restart after Redis replacement: %v", err) + } // Wait for sweep to pick up the job and push it, then worker to process. deadline := time.Now().Add(5 * time.Second) diff --git a/backend/pkg/logger/logger.go b/backend/pkg/logger/logger.go index 20983333..4b838b9c 100644 --- a/backend/pkg/logger/logger.go +++ b/backend/pkg/logger/logger.go @@ -2,6 +2,7 @@ package logger import ( "os" + "sync/atomic" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -9,7 +10,7 @@ import ( // Logger wraps zap.SugaredLogger for structured logging. // Corresponds to Chatwoot's Rails Logger but with structured output. -var globalLogger *zap.SugaredLogger +var globalLogger atomic.Pointer[zap.SugaredLogger] // Config for logger initialization type Config struct { @@ -64,25 +65,30 @@ func Init(cfg Config) error { combinedCore := zapcore.NewTee(core, errorCore) zapLogger := zap.New(combinedCore, zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel)) - globalLogger = zapLogger.Sugar() + globalLogger.Store(zapLogger.Sugar()) return nil } // L returns the global sugared logger func L() *zap.SugaredLogger { - if globalLogger == nil { - // Fallback: if Init wasn't called, use a default logger - zapLogger, _ := zap.NewProduction() - globalLogger = zapLogger.Sugar() + if logger := globalLogger.Load(); logger != nil { + return logger } - return globalLogger + + // Fallback: if Init wasn't called, use a default logger. + zapLogger, _ := zap.NewProduction() + logger := zapLogger.Sugar() + if globalLogger.CompareAndSwap(nil, logger) { + return logger + } + return globalLogger.Load() } // Sync flushes any buffered log entries. Should be called before program exit. func Sync() { - if globalLogger != nil { - _ = globalLogger.Sync() + if logger := globalLogger.Load(); logger != nil { + _ = logger.Sync() } } diff --git a/backend/pkg/logger/logger_test.go b/backend/pkg/logger/logger_test.go index 71d3cd64..8df868ba 100644 --- a/backend/pkg/logger/logger_test.go +++ b/backend/pkg/logger/logger_test.go @@ -81,10 +81,10 @@ func TestInit_JSONFormat(t *testing.T) { } err := Init(cfg) assert.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) // globalLogger should be set - require.NotNil(t, globalLogger) + require.NotNil(t, globalLogger.Load()) } func TestInit_ConsoleFormat(t *testing.T) { @@ -96,9 +96,9 @@ func TestInit_ConsoleFormat(t *testing.T) { } err := Init(cfg) assert.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) - require.NotNil(t, globalLogger) + require.NotNil(t, globalLogger.Load()) } func TestInit_FileOutput(t *testing.T) { @@ -114,9 +114,9 @@ func TestInit_FileOutput(t *testing.T) { } err := Init(cfg) assert.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) - require.NotNil(t, globalLogger) + require.NotNil(t, globalLogger.Load()) // Write a log and sync L().Info("test message from file output") @@ -138,7 +138,7 @@ func TestInit_InvalidLevel(t *testing.T) { } err := Init(cfg) assert.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) } func TestInit_InvalidOutput(t *testing.T) { @@ -165,7 +165,7 @@ func TestInit_InvalidErrorOutput(t *testing.T) { func TestL_WithoutInit(t *testing.T) { // Reset globalLogger to nil to test the fallback path - globalLogger = nil + globalLogger.Store(nil) l := L() assert.NotNil(t, l) // Should be a usable sugared logger @@ -181,7 +181,7 @@ func TestL_AfterInit(t *testing.T) { } err := Init(cfg) require.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) l := L() assert.NotNil(t, l) @@ -196,14 +196,14 @@ func TestSync_WithLogger(t *testing.T) { } err := Init(cfg) require.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) L().Info("message before sync") Sync() } func TestSync_WithoutLogger(t *testing.T) { - globalLogger = nil + globalLogger.Store(nil) // Should not panic Sync() } @@ -217,7 +217,7 @@ func TestL_LogsAtVariousLevels(t *testing.T) { } err := Init(cfg) require.NoError(t, err) - defer func() { globalLogger = nil }() + defer globalLogger.Store(nil) l := L() l.Debug("debug message")