perf(cache): 三层缓存框架补强

- 新增 cache-framework skill,规范 RAM→Redis→DB 读路径、失效与 pub/sub
- 上传元数据 Otter+Redis 缓存与多节点失效;Auth Source 缓存与 pub/sub
- ListSystemConfigsByKeys 补 Redis 层;上传统计单事务;登录/Token 缓存预热
- cleanup 任务补 upload meta 失效钩子
This commit is contained in:
ryan
2026-06-20 10:17:50 +08:00
parent 8c872f9b32
commit cdac1f8a45
23 changed files with 1511 additions and 20 deletions
+151
View File
@@ -0,0 +1,151 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package cache
import (
"context"
"encoding/json"
"fmt"
"sync"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
)
const (
uploadMetaRedisCacheTTL = 30 * 60 // seconds
uploadMetaRAMMaximumSize = 4096
uploadMetaInvalidationChan = "upload:meta_invalidation"
)
type uploadMetaInvalidationMessage struct {
ID uint64 `json:"id"`
}
var (
uploadMetaRAM = ram.MustNew[uint64, model.Upload](ram.Options{MaximumSize: uploadMetaRAMMaximumSize})
uploadMetaListenerOnce sync.Once
uploadMetaListenerCtx context.Context
uploadMetaListenerCancel context.CancelFunc
)
func uploadMetaRedisKey(id uint64) string {
return fmt.Sprintf("upload:meta:%d", id)
}
func cloneUpload(upload model.Upload) model.Upload {
return upload
}
func ensureUploadMetaCacheListener() {
if db.Redis == nil {
return
}
uploadMetaListenerOnce.Do(startUploadMetaCacheInvalidationListener)
}
func startUploadMetaCacheInvalidationListener() {
uploadMetaListenerCtx, uploadMetaListenerCancel = context.WithCancel(context.Background())
go func() {
pubsub := db.Redis.Subscribe(uploadMetaListenerCtx, uploadMetaInvalidationChan)
defer func() {
_ = pubsub.Close()
}()
go func() {
<-uploadMetaListenerCtx.Done()
_ = pubsub.Close()
}()
for msg := range pubsub.Channel() {
var payload uploadMetaInvalidationMessage
if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil || payload.ID == 0 {
uploadMetaRAM.InvalidateAll()
continue
}
uploadMetaRAM.Invalidate(payload.ID)
}
}()
}
func publishUploadMetaRAMInvalidation(ctx context.Context, id uint64) {
if db.Redis == nil {
return
}
payload, err := json.Marshal(uploadMetaInvalidationMessage{ID: id})
if err != nil {
return
}
_ = db.Redis.Publish(ctx, uploadMetaInvalidationChan, payload).Err()
}
// GetUploadByID loads upload metadata from RAM, Redis, or the database.
func GetUploadByID(ctx context.Context, id uint64) (model.Upload, error) {
ensureUploadMetaCacheListener()
if upload, ok := uploadMetaRAM.GetIfPresent(id); ok {
return cloneUpload(upload), nil
}
key := uploadMetaRedisKey(id)
if db.Redis != nil {
var upload model.Upload
if err := db.GetJSON(ctx, key, &upload); err == nil {
uploadMetaRAM.Set(id, cloneUpload(upload))
return upload, nil
}
}
var upload model.Upload
if err := db.DB(ctx).
Where("id = ? AND status IN (?, ?)", id, model.UploadStatusPending, model.UploadStatusUsed).
First(&upload).Error; err != nil {
return model.Upload{}, err
}
SetUploadMetaCache(ctx, &upload)
return upload, nil
}
// SetUploadMetaCache populates RAM and Redis upload metadata caches.
func SetUploadMetaCache(ctx context.Context, upload *model.Upload) {
ensureUploadMetaCacheListener()
if upload == nil {
return
}
cloned := cloneUpload(*upload)
uploadMetaRAM.Set(upload.ID, cloned)
if db.Redis != nil {
_ = db.SetJSON(ctx, uploadMetaRedisKey(upload.ID), cloned, uploadMetaRedisCacheTTL)
}
}
// InvalidateUploadMetaCache clears RAM and Redis upload metadata caches and notifies peer nodes.
func InvalidateUploadMetaCache(ctx context.Context, id uint64) {
ensureUploadMetaCacheListener()
uploadMetaRAM.Invalidate(id)
if db.Redis != nil {
_ = db.Redis.Del(ctx, db.PrefixedKey(uploadMetaRedisKey(id))).Err()
publishUploadMetaRAMInvalidation(ctx, id)
}
}
// ResetUploadMetaCacheForTest clears the in-process upload metadata RAM cache.
func ResetUploadMetaCacheForTest() {
uploadMetaRAM.InvalidateAll()
}
// StopUploadMetaCacheListener stops the Redis Pub/Sub subscription listener and resets the sync.Once guard.
func StopUploadMetaCacheListener() {
if uploadMetaListenerCancel != nil {
uploadMetaListenerCancel()
uploadMetaListenerCancel = nil
}
uploadMetaListenerOnce = sync.Once{}
}
+287
View File
@@ -0,0 +1,287 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package cache
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"gorm.io/gorm"
)
func init() {
testhelper.RegisterCleanup(func() {
StopUploadMetaCacheListener()
ResetUploadMetaCacheForTest()
})
}
func seedUpload(t *testing.T, dbConn *gorm.DB, upload model.Upload) {
t.Helper()
if err := dbConn.Create(&upload).Error; err != nil {
t.Fatalf("create upload: %v", err)
}
}
func TestGetUploadByIDLoadsFromDBAndPopulatesCache(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
ctx := context.Background()
upload := model.Upload{
ID: 91001,
UserID: 1,
FileName: "cached.png",
FilePath: "cached.png",
FileSize: 12,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusUsed,
AccessMode: 1,
}
seedUpload(t, dbConn, upload)
got, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID: %v", err)
}
if got.ID != upload.ID || got.FileName != upload.FileName {
t.Fatalf("unexpected upload: %+v", got)
}
var redisUpload model.Upload
if err := db.GetJSON(ctx, uploadMetaRedisKey(upload.ID), &redisUpload); err != nil {
t.Fatalf("redis cache miss after DB load: %v", err)
}
if redisUpload.ID != upload.ID {
t.Fatalf("redis upload id mismatch: got=%d want=%d", redisUpload.ID, upload.ID)
}
if err := dbConn.Delete(&model.Upload{}, upload.ID).Error; err != nil {
t.Fatalf("delete upload from db: %v", err)
}
gotCached, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID from RAM cache: %v", err)
}
if gotCached.ID != upload.ID {
t.Fatalf("expected RAM cache hit for upload %d", upload.ID)
}
}
func TestGetUploadByIDReadsFromRedisWhenRAMEmpty(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
ctx := context.Background()
upload := model.Upload{
ID: 91002,
UserID: 1,
FileName: "redis.png",
FilePath: "redis.png",
FileSize: 8,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusPending,
AccessMode: 0,
}
seedUpload(t, dbConn, upload)
SetUploadMetaCache(ctx, &upload)
ResetUploadMetaCacheForTest()
if err := dbConn.Delete(&model.Upload{}, upload.ID).Error; err != nil {
t.Fatalf("delete upload from db: %v", err)
}
got, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID from redis: %v", err)
}
if got.ID != upload.ID || got.FileName != upload.FileName {
t.Fatalf("unexpected upload from redis: %+v", got)
}
}
func TestInvalidateUploadMetaCacheClearsRAMAndRedis(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
ctx := context.Background()
upload := model.Upload{
ID: 91003,
UserID: 1,
FileName: "invalidate.png",
FilePath: "invalidate.png",
FileSize: 4,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusUsed,
AccessMode: 1,
}
seedUpload(t, dbConn, upload)
SetUploadMetaCache(ctx, &upload)
InvalidateUploadMetaCache(ctx, upload.ID)
var redisUpload model.Upload
if err := db.GetJSON(ctx, uploadMetaRedisKey(upload.ID), &redisUpload); err == nil {
t.Fatal("expected redis cache to be invalidated")
}
got, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID after invalidate should reload from DB: %v", err)
}
if got.ID != upload.ID {
t.Fatalf("unexpected upload reloaded from DB: %+v", got)
}
}
func TestUploadMetaInvalidationPubSubClearsPeerRAM(t *testing.T) {
StopUploadMetaCacheListener()
defer StopUploadMetaCacheListener()
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
ctx := context.Background()
upload := model.Upload{
ID: 91006,
UserID: 1,
FileName: "pubsub.png",
FilePath: "pubsub.png",
FileSize: 4,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusUsed,
AccessMode: 1,
}
seedUpload(t, dbConn, upload)
if _, err := GetUploadByID(ctx, upload.ID); err != nil {
t.Fatalf("GetUploadByID: %v", err)
}
time.Sleep(50 * time.Millisecond) // allow pub/sub listener to subscribe
if err := dbConn.Delete(&model.Upload{}, upload.ID).Error; err != nil {
t.Fatalf("delete upload from db: %v", err)
}
if _, err := GetUploadByID(ctx, upload.ID); err != nil {
t.Fatalf("expected cache hit before pub/sub invalidation: %v", err)
}
payload, err := json.Marshal(uploadMetaInvalidationMessage{ID: upload.ID})
if err != nil {
t.Fatalf("marshal invalidation payload: %v", err)
}
if err := db.Redis.Publish(ctx, uploadMetaInvalidationChan, string(payload)).Err(); err != nil {
t.Fatalf("publish invalidation: %v", err)
}
deadline := time.Now().Add(2 * time.Second)
ramCleared := false
for time.Now().Before(deadline) {
if _, ok := uploadMetaRAM.GetIfPresent(upload.ID); !ok {
ramCleared = true
break
}
time.Sleep(20 * time.Millisecond)
}
if !ramCleared {
t.Fatal("expected peer RAM cache to be cleared by pub/sub")
}
if err := db.Redis.Del(ctx, db.PrefixedKey(uploadMetaRedisKey(upload.ID))).Err(); err != nil {
t.Fatalf("delete redis cache: %v", err)
}
if _, err := GetUploadByID(ctx, upload.ID); err == nil {
t.Fatal("expected cache miss after pub/sub RAM eviction and redis delete")
}
}
func TestGetUploadByIDSkipsDeletedUploads(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
ctx := context.Background()
upload := model.Upload{
ID: 91004,
UserID: 1,
FileName: "deleted.png",
FilePath: "deleted.png",
FileSize: 4,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusDeleted,
AccessMode: 1,
}
seedUpload(t, dbConn, upload)
if _, err := GetUploadByID(ctx, upload.ID); err == nil {
t.Fatal("expected error for deleted upload")
}
}
func TestGetUploadByIDWorksWithRedisDisabled(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ResetUploadMetaCacheForTest()
redisClient := db.Redis
db.Redis = nil
t.Cleanup(func() {
db.Redis = redisClient
StopUploadMetaCacheListener()
})
ctx := context.Background()
upload := model.Upload{
ID: 91005,
UserID: 1,
FileName: "ram-only.png",
FilePath: "ram-only.png",
FileSize: 6,
MimeType: "image/png",
Extension: "png",
Type: "avatar",
Status: model.UploadStatusUsed,
AccessMode: 1,
}
seedUpload(t, dbConn, upload)
got, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID without redis: %v", err)
}
if got.ID != upload.ID {
t.Fatalf("unexpected upload: %+v", got)
}
if err := dbConn.Delete(&model.Upload{}, upload.ID).Error; err != nil {
t.Fatalf("delete upload from db: %v", err)
}
gotCached, err := GetUploadByID(ctx, upload.ID)
if err != nil {
t.Fatalf("GetUploadByID from RAM without redis: %v", err)
}
if gotCached.ID != upload.ID {
t.Fatal("expected RAM cache hit when redis is disabled")
}
}
+2 -5
View File
@@ -22,7 +22,6 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/upload/util"
"github.com/Rain-kl/Wavelet/internal/common"
"github.com/Rain-kl/Wavelet/internal/common/response"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/diskcache"
"github.com/Rain-kl/Wavelet/internal/model"
@@ -96,10 +95,8 @@ func GetUploadRecordByID(c *gin.Context) (*model.Upload, error) {
return nil, err
}
var upload model.Upload
if err := db.DB(c.Request.Context()).
Where("id = ? AND status IN (?, ?)", uploadID, model.UploadStatusPending, model.UploadStatusUsed).
First(&upload).Error; err != nil {
upload, err := cache.GetUploadByID(c.Request.Context(), uploadID)
if err != nil {
return nil, err
}
@@ -34,6 +34,10 @@ import (
"gorm.io/gorm"
)
func init() {
testhelper.RegisterCleanup(cache.ResetUploadMetaCacheForTest)
}
func TestServeFileByIDAccessControl(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
+13 -2
View File
@@ -11,9 +11,11 @@ import (
"strings"
"time"
uploadcache "github.com/Rain-kl/Wavelet/internal/apps/upload/cache"
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
uploadstorage "github.com/Rain-kl/Wavelet/internal/apps/upload/storage"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/db/idgen"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
@@ -99,7 +101,7 @@ func storeObject(ctx context.Context, objectKey string, reader io.Reader, size i
}
func persistUploadRecord(ctx context.Context, upload *model.Upload, objectKey string) error {
if err := repository.CreateUpload(ctx, upload); err != nil {
if err := createUploadWithStats(ctx, upload); err != nil {
_, backend, backendErr := storage.Active(ctx)
if backendErr == nil {
if deleteErr := backend.Delete(ctx, objectKey); deleteErr != nil {
@@ -108,10 +110,19 @@ func persistUploadRecord(ctx context.Context, upload *model.Upload, objectKey st
}
return err
}
uploadstats.RecordUploadStatsAdd(ctx, upload)
uploadcache.SetUploadMetaCache(ctx, upload)
return nil
}
func createUploadWithStats(ctx context.Context, upload *model.Upload) error {
return db.DB(ctx).Transaction(func(tx *gorm.DB) error {
if err := repository.CreateUploadTx(tx, upload); err != nil {
return err
}
return uploadstats.ApplyUploadStatsDeltaTx(tx, upload, 1)
})
}
func createDedupRecord(ctx context.Context, existing model.Upload, req Request) (Result, error) {
accessMode := resolveAccessMode(req.Type, req.AccessMode)
newUpload := model.Upload{
@@ -183,6 +183,52 @@ func TestIngestPolicyDedupNewRecordCreatesSecondRecord(t *testing.T) {
}
}
func TestCreateUploadWithStatsRollsBackOnCreateFailure(t *testing.T) {
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ctx := context.Background()
existing := model.Upload{
ID: 99001,
UserID: 1001,
FileName: "existing.png",
FilePath: "uploads/existing.png",
FileSize: 64,
MimeType: "image/png",
Extension: "png",
Type: "generic",
Status: model.UploadStatusUsed,
CreatedAt: time.Now(),
}
if err := dbConn.Create(&existing).Error; err != nil {
t.Fatalf("seed upload failed: %v", err)
}
duplicate := &model.Upload{
ID: existing.ID,
UserID: 1002,
FileName: "duplicate.png",
FilePath: "uploads/duplicate.png",
FileSize: 128,
MimeType: "image/png",
Extension: "png",
Type: "generic",
Status: model.UploadStatusUsed,
CreatedAt: time.Now(),
}
if err := createUploadWithStats(ctx, duplicate); err == nil {
t.Fatal("createUploadWithStats with duplicate ID expected error")
}
stats, err := loadTotalStats(ctx)
if err != nil {
t.Fatalf("loadTotalStats returned error: %v", err)
}
if stats.TotalCount != 0 || stats.TotalSize != 0 {
t.Fatalf("loadTotalStats() = count %d size %d, want zero after rolled-back stats", stats.TotalCount, stats.TotalSize)
}
}
func TestRemoveDecrementsStats(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
+19 -4
View File
@@ -6,9 +6,12 @@ package ingest
import (
"context"
uploadcache "github.com/Rain-kl/Wavelet/internal/apps/upload/cache"
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
"gorm.io/gorm"
)
// Remove soft-deletes an upload and decrements incremental stats.
@@ -17,8 +20,7 @@ func Remove(ctx context.Context, uploadID uint64) (model.Upload, error) {
if err != nil {
return model.Upload{}, err
}
uploadstats.RecordUploadStatsRemove(ctx, &upload)
if err := repository.SoftDeleteUpload(ctx, &upload); err != nil {
if err := softDeleteUploadWithStats(ctx, &upload); err != nil {
return model.Upload{}, err
}
upload.Status = model.UploadStatusDeleted
@@ -34,10 +36,23 @@ func RemoveOwned(ctx context.Context, userID, uploadID uint64) (model.Upload, er
if upload.UserID != userID {
return model.Upload{}, ErrForbidden
}
uploadstats.RecordUploadStatsRemove(ctx, &upload)
if err := repository.SoftDeleteUpload(ctx, &upload); err != nil {
if err := softDeleteUploadWithStats(ctx, &upload); err != nil {
return model.Upload{}, err
}
upload.Status = model.UploadStatusDeleted
return upload, nil
}
func softDeleteUploadWithStats(ctx context.Context, upload *model.Upload) error {
statsSnapshot := *upload
if err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
if err := repository.SoftDeleteUploadTx(tx, upload); err != nil {
return err
}
return uploadstats.ApplyUploadStatsDeltaTx(tx, &statsSnapshot, -1)
}); err != nil {
return err
}
uploadcache.InvalidateUploadMetaCache(ctx, upload.ID)
return nil
}
+4 -3
View File
@@ -37,7 +37,7 @@ func RebuildUploadStats(ctx context.Context) error {
}
for i := range uploads {
if err := applyUploadStatsDeltaTx(tx, &uploads[i], 1); err != nil {
if err := ApplyUploadStatsDeltaTx(tx, &uploads[i], 1); err != nil {
return err
}
}
@@ -50,11 +50,12 @@ func applyUploadStatsDelta(ctx context.Context, upload *model.Upload, sign int64
return nil
}
return db.DB(ctx).Transaction(func(tx *gorm.DB) error {
return applyUploadStatsDeltaTx(tx, upload, sign)
return ApplyUploadStatsDeltaTx(tx, upload, sign)
})
}
func applyUploadStatsDeltaTx(tx *gorm.DB, upload *model.Upload, sign int64) error {
// ApplyUploadStatsDeltaTx applies incremental upload stats within an existing transaction.
func ApplyUploadStatsDeltaTx(tx *gorm.DB, upload *model.Upload, sign int64) error {
if upload == nil || !isActiveUploadStatus(upload.Status) || sign == 0 {
return nil
}
@@ -11,8 +11,39 @@ import (
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"gorm.io/gorm"
)
func TestApplyUploadStatsDeltaTxWithinTransaction(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
ctx := context.Background()
upload := &model.Upload{
ID: 42002,
FileSize: 256,
MimeType: "image/jpeg",
Extension: "jpg",
Type: "avatar",
Status: model.UploadStatusUsed,
CreatedAt: time.Now(),
}
if err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
return ApplyUploadStatsDeltaTx(tx, upload, 1)
}); err != nil {
t.Fatalf("ApplyUploadStatsDeltaTx returned error: %v", err)
}
stats, err := loadUploadStats(ctx)
if err != nil {
t.Fatalf("loadUploadStats returned error: %v", err)
}
if stats.TotalCount != 1 || stats.TotalSize != 256 {
t.Fatalf("unexpected total stats: count=%d size=%d", stats.TotalCount, stats.TotalSize)
}
}
func TestApplyUploadStatsAddAndRemove(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
+2
View File
@@ -10,6 +10,7 @@ import (
"fmt"
"time"
uploadcache "github.com/Rain-kl/Wavelet/internal/apps/upload/cache"
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
uploadstorage "github.com/Rain-kl/Wavelet/internal/apps/upload/storage"
@@ -100,6 +101,7 @@ func (h *SystemCleanupHandler) Execute(ctx context.Context, _ []byte) (*task.Tas
}
uploadstats.RecordUploadStatsRemove(ctx, &u)
uploadcache.InvalidateUploadMetaCache(ctx, u.ID)
totalDeleted++
lastID = u.ID
}