From b179ec6de7a361b6c8843b23a31a2c3527cb9b8b Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sun, 11 Oct 2026 08:30:58 +0300 Subject: [PATCH] =?UTF-8?q?persistence:=20Client=20=E2=86=92=20atomic.Poin?= =?UTF-8?q?ter=20(=D1=83=D1=81=D1=82=D1=80=D0=B0=D0=BD=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D0=B5=20data=20race=20=D0=BF=D1=80=D0=B8=20reconnect)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/cmd/goaws.go | 4 +- app/persistence/redis.go | 93 +++++++++++++++++++++------------------- 2 files changed, 50 insertions(+), 47 deletions(-) diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index 948cea6..86abb4e 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -109,7 +109,7 @@ func main() { } // Восстановление состояния из Redis (тенанты + очереди) - if persistence.Client != nil { + if persistence.Client.Load() != nil { // Загружаем тенантов tenantsRaw, err := persistence.LoadAllTenantsRaw() if err != nil { @@ -139,7 +139,7 @@ func main() { // Бэкфилл индекса accessKey→tenant для read-through (идемпотентно; нужен для // тенантов, созданных до появления индекса, и для multi-instance сценария). - if persistence.Client != nil { + if persistence.Client.Load() != nil { for _, t := range tenantStore.List() { if data, err := json.Marshal(t); err == nil { persistence.SaveTenantRaw(t.ID, data) diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 5e120c9..8de0839 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -20,6 +20,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" "github.com/redis/go-redis/v9" @@ -30,7 +31,9 @@ import ( // Client — глобальный Redis клиент. // nil означает режим "только память" — все функции тихо no-op. -var Client *redis.Client +// atomic.Pointer: reconnect() пишет новый клиент из health-check горутины, +// а остальные горутины читают — без атомарности это data race. +var Client atomic.Pointer[redis.Client] // Параметры подключения — сохраняются для переподключения после обрыва. var ( @@ -78,7 +81,7 @@ func Connect(addr, username, password string) error { if err := rdb.Ping(ctx).Err(); err != nil { return fmt.Errorf("redis ping %s: %w", addr, err) } - Client = rdb + Client.Store(rdb) log.Infof("persistence: подключились к Redis %s", addr) startRedisHealthCheck() return nil @@ -93,11 +96,11 @@ func startRedisHealthCheck() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for range ticker.C { - if Client == nil { + if Client.Load() == nil { continue } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - err := Client.Ping(ctx).Err() + err := Client.Load().Ping(ctx).Err() cancel() if err == nil { continue @@ -128,7 +131,7 @@ func reconnect() error { if err := rdb.Ping(ctx).Err(); err != nil { return err } - Client = rdb + Client.Store(rdb) return nil } @@ -222,7 +225,7 @@ func marshalQueueMeta(queue *models.Queue) ([]byte, error) { // создаёт консистентный снапшот. Горутина только делает сетевой вызов. // Для сохранения сообщений используй SaveMessage/SaveMessages. func SaveQueue(key string, queue *models.Queue) { - if Client == nil { + if Client.Load() == nil { return } // Сериализуем метаданные синхронно под мьютексом вызывающего → консистентный снапшот @@ -238,7 +241,7 @@ func SaveQueue(key string, queue *models.Queue) { } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HSet(ctx, redisHashQueues, key, string(data)).Err(); err != nil { + if err := Client.Load().HSet(ctx, redisHashQueues, key, string(data)).Err(); err != nil { log.Errorf("persistence: HSet queue meta %q: %v", key, err) } }) @@ -248,7 +251,7 @@ func SaveQueue(key string, queue *models.Queue) { // Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage). // Вызывать под Lock — json.Marshal делает консистентный снапшот сообщения. func SaveMessage(queueKey string, msg *models.SqsMessage) { - if Client == nil { + if Client.Load() == nil { return } data, err := json.Marshal(msg) @@ -260,7 +263,7 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) { asyncWrite(func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil { + if err := Client.Load().HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil { log.Errorf("persistence: HSet message %s/%s: %v", queueKey, msg.Uuid, err) } }) @@ -273,7 +276,7 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) { // Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage). // Если Client == nil (memory-only режим) — возвращает nil (персистить некуда). func SaveMessageSync(queueKey string, msg *models.SqsMessage) error { - if Client == nil { + if Client.Load() == nil { return nil } data, err := json.Marshal(msg) @@ -283,7 +286,7 @@ func SaveMessageSync(queueKey string, msg *models.SqsMessage) error { hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil { + if err := Client.Load().HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil { return fmt.Errorf("HSet message %s/%s: %w", queueKey, msg.Uuid, err) } return nil @@ -292,7 +295,7 @@ func SaveMessageSync(queueKey string, msg *models.SqsMessage) error { // SaveMessages — сохраняет несколько сообщений в Redis одним pipeline. // Вызывать под Lock. func SaveMessages(queueKey string, msgs []models.SqsMessage) { - if Client == nil || len(msgs) == 0 { + if Client.Load() == nil || len(msgs) == 0 { return } // Сериализуем все сообщения синхронно под Lock @@ -317,7 +320,7 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) { for k, v := range fields { args = append(args, k, v) } - if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil { + if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil { log.Errorf("persistence: HSet messages %s (%d msgs): %v", queueKey, len(fields), err) } }) @@ -327,7 +330,7 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) { // Используется на пути SendMessageBatch: ответ 200 выдаётся только после записи // всех сообщений (как в одиночном SendMessage). Возвращает ошибку для отката. func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error { - if Client == nil || len(msgs) == 0 { + if Client.Load() == nil || len(msgs) == 0 { return nil } fields := make(map[string]string, len(msgs)) @@ -348,7 +351,7 @@ func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error { for k, v := range fields { args = append(args, k, v) } - if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil { + if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil { return fmt.Errorf("HSet messages %s (%d msgs): %w", queueKey, len(fields), err) } return nil @@ -359,13 +362,13 @@ func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error { // сообщение «воскреснет» из Redis (факт: удалённые очереди возвращались // после рестарта, когда удаления были асинхронными). func DeleteMessagePersist(queueKey string, msgUuid string) { - if Client == nil || msgUuid == "" { + if Client.Load() == nil || msgUuid == "" { return } hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HDel(ctx, hashKey, msgUuid).Err(); err != nil { + if err := Client.Load().HDel(ctx, hashKey, msgUuid).Err(); err != nil { log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err) } } @@ -373,13 +376,13 @@ func DeleteMessagePersist(queueKey string, msgUuid string) { // DeleteMessagesPersist — удаляет несколько сообщений из Redis СИНХРОННО // (иначе удалённые сообщения воскреснут после рестарта пода). func DeleteMessagesPersist(queueKey string, uuids []string) { - if Client == nil || len(uuids) == 0 { + if Client.Load() == nil || len(uuids) == 0 { return } hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - if err := Client.HDel(ctx, hashKey, uuids...).Err(); err != nil { + if err := Client.Load().HDel(ctx, hashKey, uuids...).Err(); err != nil { log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(uuids), err) } } @@ -388,20 +391,20 @@ func DeleteMessagesPersist(queueKey string, uuids []string) { // Синхронность обязательна: при рестарте пода удалённые сообщения не должны // воскреснуть (факт: асинхронное удаление очередей приводило к их возврату). func PurgeMessagesPersist(queueKey string) { - if Client == nil { + if Client.Load() == nil { return } hashKey := redisMsgHashPrefix + queueKey ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - if err := Client.Del(ctx, hashKey).Err(); err != nil { + if err := Client.Load().Del(ctx, hashKey).Err(); err != nil { log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err) } } // DeleteQueue — удаляет метаданные очереди И все её сообщения из Redis асинхронно. func DeleteQueue(key string) { - if Client == nil { + if Client.Load() == nil { return } writeSeq := nextQueueWriteSeq(key) @@ -415,11 +418,11 @@ func DeleteQueue(key string) { return } // Удаляем метаданные из общего HASH - if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil { + if err := Client.Load().HDel(ctx, redisHashQueues, key).Err(); err != nil { log.Errorf("persistence: HDel queue meta %q: %v", key, err) } // Удаляем весь HASH с сообщениями - if err := Client.Del(ctx, msgHashKey).Err(); err != nil { + if err := Client.Load().Del(ctx, msgHashKey).Err(); err != nil { log.Errorf("persistence: DEL messages hash %q: %v", key, err) } } @@ -429,14 +432,14 @@ func DeleteQueue(key string) { // Миграция v1→v2: если в ssq:queues лежит JSON со встроенными Messages, // они извлекаются в отдельный HASH и метаданные пересохраняются без Messages. func LoadAllQueues() (map[string]*models.Queue, error) { - if Client == nil { + if Client.Load() == nil { return nil, nil } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() // 1. Загружаем метаданные очередей - raw, err := Client.HGetAll(ctx, redisHashQueues).Result() + raw, err := Client.Load().HGetAll(ctx, redisHashQueues).Result() if err != nil { return nil, fmt.Errorf("redis HGetAll queues: %w", err) } @@ -496,7 +499,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) { args = append(args, q.Messages[i].Uuid, string(data)) } if len(args) > 0 { - if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil { + if err := Client.Load().HSet(ctx, hashKey, args...).Err(); err != nil { log.Errorf("persistence: migrate HSet messages %q: %v", k, err) } } @@ -504,7 +507,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) { // Пересохраняем метаданные без Messages metaData, err := marshalQueueMeta(q) if err == nil { - if err := Client.HSet(ctx, redisHashQueues, k, string(metaData)).Err(); err != nil { + if err := Client.Load().HSet(ctx, redisHashQueues, k, string(metaData)).Err(); err != nil { log.Errorf("persistence: migrate HSet meta %q: %v", k, err) } } @@ -514,7 +517,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) { // 3. Загружаем сообщения из отдельных HASH'ов для каждой очереди for k, q := range queues { hashKey := redisMsgHashPrefix + k - msgRaw, err := Client.HGetAll(ctx, hashKey).Result() + msgRaw, err := Client.Load().HGetAll(ctx, hashKey).Result() if err != nil { log.Errorf("persistence: HGetAll messages %q: %v", k, err) continue @@ -565,7 +568,7 @@ func LoadAllQueues() (map[string]*models.Queue, error) { // Принимает []byte чтобы избежать циклического импорта с пакетом tenant. // Сериализацию делает вызывающий (tenant_store.go). func SaveTenantRaw(id string, jsonData []byte) { - if Client == nil { + if Client.Load() == nil { return } // Копируем bytes — вызывающий может переиспользовать буфер @@ -583,11 +586,11 @@ func SaveTenantRaw(id string, jsonData []byte) { } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil { + if err := Client.Load().HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil { log.Errorf("persistence: HSet tenant %q: %v", id, err) } if meta.AccessKey != "" { - if err := Client.HSet(ctx, redisHashTenantAK, meta.AccessKey, string(dataCopy)).Err(); err != nil { + if err := Client.Load().HSet(ctx, redisHashTenantAK, meta.AccessKey, string(dataCopy)).Err(); err != nil { log.Errorf("persistence: HSet tenant_ak %q: %v", meta.AccessKey, err) } } @@ -597,12 +600,12 @@ func SaveTenantRaw(id string, jsonData []byte) { // LoadTenantByAccessKey — read-through тенанта по accessKey (для auth-middleware). // Возвращает сырой JSON; десериализацию делает вызывающий. func LoadTenantByAccessKey(accessKey string) ([]byte, bool) { - if Client == nil || accessKey == "" { + if Client.Load() == nil || accessKey == "" { return nil, false } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - v, err := Client.HGet(ctx, redisHashTenantAK, accessKey).Result() + v, err := Client.Load().HGet(ctx, redisHashTenantAK, accessKey).Result() if err != nil { return nil, false } @@ -613,7 +616,7 @@ func LoadTenantByAccessKey(accessKey string) ([]byte, bool) { // Удаляет из ОБОИХ индексов: ssq:tenants (по id) и ssq:tenant_ak (по accessKey), // иначе LoadTenantByAccessKey вернёт удалённого тенанта и отзыв ключей не сработает. func DeleteTenant(id, accessKey string) { - if Client == nil { + if Client.Load() == nil { return } writeSeq := nextTenantWriteSeq(id) @@ -623,11 +626,11 @@ func DeleteTenant(id, accessKey string) { } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if err := Client.HDel(ctx, redisHashTenants, id).Err(); err != nil { + if err := Client.Load().HDel(ctx, redisHashTenants, id).Err(); err != nil { log.Errorf("persistence: HDel tenant %q: %v", id, err) } if accessKey != "" { - if err := Client.HDel(ctx, redisHashTenantAK, accessKey).Err(); err != nil { + if err := Client.Load().HDel(ctx, redisHashTenantAK, accessKey).Err(); err != nil { log.Errorf("persistence: HDel tenant_ak %q: %v", accessKey, err) } } @@ -663,12 +666,12 @@ func isLatestTenantWriteSeq(id string, seq uint64) bool { // LoadAllTenantsRaw — загружает всех тенантов из Redis при старте. // Возвращает map[tenantID]rawJSON — десериализацию делает tenant_store.go. func LoadAllTenantsRaw() (map[string][]byte, error) { - if Client == nil { + if Client.Load() == nil { return nil, nil } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - raw, err := Client.HGetAll(ctx, redisHashTenants).Result() + raw, err := Client.Load().HGetAll(ctx, redisHashTenants).Result() if err != nil { return nil, fmt.Errorf("redis HGetAll tenants: %w", err) } @@ -690,12 +693,12 @@ func LoadAllTenantsRaw() (map[string][]byte, error) { // TryAcquireLock — пытается взять лок (SETNX с TTL). true — лок наш. func TryAcquireLock(key, value string, ttl time.Duration) (bool, error) { - if Client == nil { + if Client.Load() == nil { return true, nil } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - return Client.SetNX(ctx, key, value, ttl).Result() + return Client.Load().SetNX(ctx, key, value, ttl).Result() } // refreshLockScript — продлить TTL, ТОЛЬКО если лок всё ещё наш (атомарно). @@ -716,12 +719,12 @@ end`) // RefreshLock — продлевает TTL лока, если он всё ещё наш. false — лок потерян. func RefreshLock(key, value string, ttl time.Duration) (bool, error) { - if Client == nil { + if Client.Load() == nil { return true, nil } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - n, err := refreshLockScript.Run(ctx, Client, []string{key}, value, ttl.Milliseconds()).Int() + n, err := refreshLockScript.Run(ctx, Client.Load(), []string{key}, value, ttl.Milliseconds()).Int() if err != nil { return false, err } @@ -730,12 +733,12 @@ func RefreshLock(key, value string, ttl time.Duration) (bool, error) { // ReleaseLock — освобождает лок (если он наш). Вызывается при graceful shutdown. func ReleaseLock(key, value string) { - if Client == nil { + if Client.Load() == nil { return } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - if _, err := releaseLockScript.Run(ctx, Client, []string{key}, value).Int(); err != nil { + if _, err := releaseLockScript.Run(ctx, Client.Load(), []string{key}, value).Int(); err != nil { log.Errorf("persistence: release lock %q: %v", key, err) } }