Nail/SQS-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

persistence: Client → atomic.Pointer (устранение data race при reconnect)

Browse Source
This commit is contained in:
Repinoid
2026-10-11 08:30:58 +03:00
parent 9e0ad2c0a4
commit b179ec6de7
2 changed files with 50 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/cmd/goaws.go
+2 -2
View File
@@ -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)
app/persistence/redis.go
+48 -45
View File
@@ -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)
}
}