feat(storage): implement Redis distributed lock and cache invalidation broadcasting

- Add Redis distributed lock in Execute of MigrationHandler to prevent concurrent storage migrations.
- Define storage:config_invalidation Redis pub/sub channel to broadcast cache invalidation events.
- Implement background pub/sub listener on all nodes to evict config memory cache concurrently.
- Add integration tests for distributed locking and invalidation propagation using miniredis.
This commit is contained in:
ryan
2026-06-13 15:21:32 +08:00
parent acd430836f
commit 26a1e71a27
6 changed files with 187 additions and 5 deletions
@@ -248,6 +248,7 @@ func UpdateSystemConfig(c *gin.Context) {
if key == model.ConfigKeyStorageConfig {
storage.ResetCache()
storage.PublishCacheInvalidation(c.Request.Context())
}
c.JSON(http.StatusOK, util.OKNil())
@@ -14,6 +14,7 @@ import (
"os"
"strings"
"sync/atomic"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -67,6 +68,18 @@ func (h *MigrationHandler) ValidatePayload(payload []byte) ([]byte, error) {
// Execute migrates all unique active-storage objects to the pending backend.
func (h *MigrationHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
if db.Redis != nil {
lockKey := db.PrefixedKey("lock:storage:migrate")
ok, err := db.Redis.SetNX(ctx, lockKey, "locked", time.Hour).Result()
if err != nil {
return nil, fmt.Errorf("acquire migration lock: %w", err)
}
if !ok {
return nil, errors.New("另一个存储迁移任务正在运行中")
}
defer db.Redis.Del(ctx, lockKey)
}
active, err := storage.LoadConfig(ctx)
if err != nil {
return nil, fmt.Errorf("load active storage config: %w", err)
@@ -14,10 +14,14 @@ import (
"path/filepath"
"strings"
"testing"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/storage"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/alicebob/miniredis/v2"
"github.com/redis/go-redis/v9"
)
func TestMigrationHandlerExecute(t *testing.T) {
@@ -220,3 +224,62 @@ func TestMigrationHandlerExecuteWithHashValidation(t *testing.T) {
t.Errorf("StorageDriver = %q, want %q", migrated.StorageDriver, storage.DriverS3)
}
}
func TestMigrationHandlerExecuteWithRedisLock(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
mr, err := miniredis.Run()
if err != nil {
t.Fatalf("Failed to run miniredis: %v", err)
}
defer mr.Close()
rdb := redis.NewClient(&redis.Options{
Addr: mr.Addr(),
})
defer rdb.Close()
oldRedis := db.Redis
db.Redis = rdb
defer func() {
db.Redis = oldRedis
}()
ctx := context.Background()
// Acquire lock manually
lockKey := db.PrefixedKey("lock:storage:migrate")
if err := rdb.Set(ctx, lockKey, "locked", time.Hour).Err(); err != nil {
t.Fatalf("Failed to set manual lock in Redis: %v", err)
}
active := storage.DefaultConfig()
if err := storage.SaveActiveConfig(ctx, active); err != nil {
t.Fatalf("SaveActiveConfig() returned error: %v", err)
}
payload, err := json.Marshal(storageMigrationPayload{Target: active})
if err != nil {
t.Fatalf("Marshal payload failed: %v", err)
}
// Execution should fail because lock is already acquired
_, err = (&MigrationHandler{}).Execute(ctx, payload)
if err == nil {
t.Fatal("Execute() succeeded when lock was held, want error")
}
if !strings.Contains(err.Error(), "另一个存储迁移任务正在运行中") {
t.Errorf("expected lock warning, got: %v", err)
}
// Release lock and run again, should succeed
if err := rdb.Del(ctx, lockKey).Err(); err != nil {
t.Fatalf("Failed to delete lock: %v", err)
}
_, err = (&MigrationHandler{}).Execute(ctx, payload)
if err != nil {
t.Fatalf("Execute() failed after lock released: %v", err)
}
}