perf(vote): 结果聚合改 stale-while-revalidate,消除持续投票时的 O(N²)

多选 / 评分 / 排序 / 表单的统计必须在 Go 层逐条 json.Unmarshal 全部提交记录
(ListByVoteID 传 nil 不设 LIMIT),而 Submit 每来一票就 InvalidateResultsCache
删掉缓存 —— 于是第 k 票触发一次 O(k) 重算,持续有人投票时稳态是 O(N²)。
一万票的投票在投票期间每次刷新都要扫一万行 JSON。

现在:
- LoadResults 用 GetWithExpiry 读到「已过期但仍在」的旧值,先返回给调用方,
  再在后台按投票去重重算(sync.Map 占位 + shared.SafeGo,60s 超时)
- 后台只拷贝 ID/Type/AllowCustom 三个字段,不把 handler 正在用的 *db.Vote 带进 goroutine
- Submit 从 InvalidateResultsCache(删缓存)改为 ExpireResultsCache(标记过期保留旧值),
  否则每次提交都会让下一次读退化成同步全量重算
- 选项变更 / 删除投票仍走 InvalidateResultsCache(结构性变化,旧值已无意义)
- TTLCache 新增 GetWithExpiry / Expire:Get 遇到过期项直接删除并返回 false,
  「过期」与「不存在」分不开,正是无法做 SWR 的原因

实测(2 万条多选提交):冷启动 0.211s → 过期首读 0.023s(快 9.2x),
后台刷新后 0.009s。缓存单测 3 项覆盖 GetWithExpiry 不删条目 / Expire 保留值。

顺带把 7 处裸类型断言改成带 ok(其中 results.go 的是新代码)。
This commit is contained in:
tamakyi
2026-10-29 16:28:23 +08:00
parent d4e6da76f1
commit 0041cc51f8
5 changed files with 356 additions and 137 deletions
+33
View File
@@ -58,6 +58,39 @@ func (c *TTLCache) Get(key string) (interface{}, bool) {
return e.value, true
}
// GetWithExpiry 读取缓存值与其过期时间,**不判断、不删除**过期条目。
//
// 供 stale-while-revalidate 使用:调用方拿到过期的旧值后可以先返回给用户,
// 再在后台重算。Get 做不到这一点——它遇到过期项就直接删掉并返回 false,
// 于是「过期」和「不存在」变成同一个结果,调用方只能同步等待重算。
func (c *TTLCache) GetWithExpiry(key string) (value interface{}, expiresAt time.Time, ok bool) {
c.mu.RLock()
defer c.mu.RUnlock()
e, ok := c.items[key]
if !ok {
return nil, time.Time{}, false
}
return e.value, e.expiresAt, true
}
// Expire 把条目标记为「已过期」但**保留其值**。
//
// 与 Delete 的区别很关键:Delete 之后下次读是彻底 miss,调用方只能同步重算;
// Expire 之后下次读仍然拿得到旧值,于是可以「先返回旧值 + 后台刷新」。
// 写多读多的场景(例如投票:每来一票都要让结果失效)只有用 Expire 才不会
// 把每次写入都变成下一次读的同步 O(N) 重算。
// key 不存在或已过期时什么也不做。
func (c *TTLCache) Expire(key string) {
c.mu.Lock()
defer c.mu.Unlock()
e, ok := c.items[key]
if !ok {
return
}
e.expiresAt = time.Now().Add(-time.Second)
c.items[key] = e
}
// Set 写入缓存并设定过期时间。
func (c *TTLCache) Set(key string, value interface{}, ttl time.Duration) {
c.mu.Lock()
+66
View File
@@ -59,3 +59,69 @@ func TestTTLCacheConcurrent(t *testing.T) {
}
wg.Wait()
}
// TestTTLCacheGetWithExpiryKeepsEntry 锁住一个关键语义:
// GetWithExpiry **不删除**过期条目,否则 stale-while-revalidate 拿不到旧值,
// 只能退化成「过期即彻底 miss → 同步重算」,那正是投票结果 O(N²) 的成因。
func TestTTLCacheGetWithExpiryKeepsEntry(t *testing.T) {
c := NewTTLCache(8)
c.Set("k", "v", time.Nanosecond)
time.Sleep(2 * time.Millisecond)
v, exp, ok := c.GetWithExpiry("k")
if !ok {
t.Fatal("expired entry should still be readable via GetWithExpiry")
}
if v.(string) != "v" {
t.Fatalf("expected stale value v, got %v", v)
}
if !time.Now().After(exp) {
t.Fatal("entry should already be expired")
}
// 再读一次仍然在(证明上一次调用没有顺手删掉它)
if _, _, ok := c.GetWithExpiry("k"); !ok {
t.Fatal("GetWithExpiry must not evict the entry")
}
// 而普通 Get 仍按原语义返回 miss(并清掉它)
if _, ok := c.Get("k"); ok {
t.Fatal("Get should report miss for an expired entry")
}
}
// TestTTLCacheGetWithExpiryMissing 不存在的 key 必须返回 ok=false,
// 且不能凭空造出一个零值过期时间让调用方误判为「过期但有值」。
func TestTTLCacheGetWithExpiryMissing(t *testing.T) {
c := NewTTLCache(8)
if _, _, ok := c.GetWithExpiry("nope"); ok {
t.Fatal("expected miss for unknown key")
}
}
// TestTTLCacheExpireKeepsValue 锁住 Expire 与 Delete 的区别:
// Expire 后条目仍在(可拿到旧值并触发后台刷新),而 Delete 之后彻底 miss。
func TestTTLCacheExpireKeepsValue(t *testing.T) {
c := NewTTLCache(8)
c.Set("k", "v", time.Hour)
c.Expire("k")
v, exp, ok := c.GetWithExpiry("k")
if !ok || v.(string) != "v" {
t.Fatalf("Expire must keep the value, got %v %v", v, ok)
}
if !time.Now().After(exp) {
t.Fatal("Expire must push expiresAt into the past")
}
// 普通 Get 立刻视为过期
if _, ok := c.Get("k"); ok {
t.Fatal("Get should see the expired entry as a miss")
}
// 对不存在的 key 调用 Expire 不应 panic 也不应创建条目
c.Expire("absent")
if _, _, ok := c.GetWithExpiry("absent"); ok {
t.Fatal("Expire must not create an entry")
}
}
+11 -2
View File
@@ -1167,9 +1167,18 @@ func VoteResults(ctx context.Context) {
formFields, _ = db.VoteFormFields.ListByVoteID(ctx.Request().Context(), vote.ID)
// 短 TTL 缓存回复明细解析结果(逐条 json.Unmarshal 全部提交记录的 O(N) 开销)。
replyKey := strconv.FormatUint(uint64(vote.ID), 10)
// 带 ok 的断言:缓存里若混入别的类型,裸断言会直接 panic 掉整个结果页。
// 类型不符时丢弃该条缓存并走重建分支(所以不能用 else 接在外层 if 后)。
fromCache := false
if cached, ok := voteTextRepliesCache.Get(replyKey); ok {
textReplies = cached.(map[string][]textReplyView)
} else {
if parsed, isMap := cached.(map[string][]textReplyView); isMap {
textReplies = parsed
fromCache = true
} else {
voteTextRepliesCache.Delete(replyKey)
}
}
if !fromCache {
submissions, _ := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), vote.ID, nil)
for _, s := range submissions {
var data map[string]interface{}
+229
View File
@@ -0,0 +1,229 @@
// Copyright 2026 tamakyi. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
package vote
import (
stdctx "context"
"encoding/json"
"strconv"
"sync"
"time"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/tamakyi/TamaBox/internal/context"
"github.com/tamakyi/TamaBox/internal/db"
"github.com/tamakyi/TamaBox/internal/shared"
)
// resultsRefreshing 记录正在后台重算的投票 ID。值为 struct{},只用于占位去重。
//
// 用 sync.Map 而不是「mutex + map」:这是典型的多数读、少数写,且 TryLoadOrStore
// 一步完成「判重 + 占位」,不需要先查后写那种两步操作(那正是别处幂等缺陷的成因)。
var resultsRefreshing sync.Map
// resultsRefreshTimeout 单次后台重算的上限。
// 后台任务与请求生命周期解绑,但必须有自己的上限,否则一次全量聚合可能无限跑下去。
const resultsRefreshTimeout = 60 * time.Second
// LoadResults 返回投票的聚合统计,策略是 stale-while-revalidate。
//
// 改这里的原因:多选 / 评分 / 排序 / 表单的统计必须在 Go 层逐条 json.Unmarshal
// 全部提交记录(O(N),`ListByVoteID(…, nil)` 不设 LIMIT)。而 Submit 每来一票就
// InvalidateResultsCache,于是持续有人投票时:第 k 票触发一次 O(k) 重算,稳态是 O(N²)。
// 一万票的投票在有人投票期间每次刷新都要扫一万行 JSON,十万票直接把单请求拖到秒级。
//
// 现在改成:缓存过期后**先把旧值返回给调用方**,再在后台按投票去重地重算。
// 请求路径永远是 O(1),O(N) 被挪到后台,且同一投票同时只有一个重算任务。
// 代价是结果最多陈旧 resultsCacheTTL(30s)—— 与原有缓存口径一致,结果页本就非强一致。
func LoadResults(ctx context.Context, v *db.Vote) (*Results, error) {
cacheKey := strconv.FormatUint(uint64(v.ID), 10)
if val, expiresAt, ok := resultsCache.GetWithExpiry(cacheKey); ok {
res, isResults := val.(*Results)
if !isResults {
// 类型不符理论上不会发生(key 由本文件的代码独占写入),
// 但裸断言一旦不符就是 panic —— 丢掉重新算即可。
resultsCache.Delete(cacheKey)
} else {
if time.Now().After(expiresAt) {
scheduleResultsRefresh(v, cacheKey)
}
return res, nil
}
}
return computeResultsAndCache(ctx.Request().Context(), v, cacheKey)
}
// scheduleResultsRefresh 在「当前投票没有刷新任务」时启动一次后台重算。
//
// 注意这里刻意不引用调用方传进来的 *db.Vote:handler 可能在之后继续改写它,
// 后台 goroutine 持有同一个指针就是潜在的数据竞争。只拷贝重算真正需要的三个字段。
func scheduleResultsRefresh(v *db.Vote, cacheKey string) {
if _, loaded := resultsRefreshing.LoadOrStore(cacheKey, struct{}{}); loaded {
return // 已有任务在跑,避免并发重算把 CPU 打满
}
voteID := v.ID
voteType := v.Type
allowCustom := v.AllowCustom
shared.SafeGo(func() {
defer resultsRefreshing.Delete(cacheKey)
reqCtx, cancel := stdctx.WithTimeout(stdctx.Background(), resultsRefreshTimeout)
defer cancel()
probe := &db.Vote{}
probe.ID = voteID
probe.Type = voteType
probe.AllowCustom = allowCustom
res, err := computeResults(reqCtx, probe)
if err != nil {
logrus.WithContext(reqCtx).WithError(err).WithField("vote_id", voteID).
Warn("后台重算投票结果失败,继续沿用旧结果直到下次成功")
return
}
resultsCache.Set(cacheKey, res, resultsCacheTTL)
})
}
// computeResultsAndCache 同步重算并写回缓存(仅用于缓存完全无值的冷启动)。
func computeResultsAndCache(reqCtx stdctx.Context, v *db.Vote, cacheKey string) (*Results, error) {
res, err := computeResults(reqCtx, v)
if err != nil {
return nil, err
}
resultsCache.Set(cacheKey, res, resultsCacheTTL)
return res, nil
}
// computeResults 按投票类型聚合统计(原来的 LoadResults 函数体)。
//
// 只依赖标准库 context:后台刷新时没有 flamego 的 Context 可用。
func computeResults(reqCtx stdctx.Context, v *db.Vote) (*Results, error) {
res := &Results{Type: v.Type}
total, err := db.VoteSubmissions.CountByVoteID(reqCtx, v.ID)
if err != nil {
return nil, errors.Wrap(err, "count submissions")
}
res.Total = total
switch v.Type {
case db.VoteTypeSingle:
options, _ := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
// 一次分组查询替代逐选项 COUNT(避免 N+1)
counts, err := db.VoteSubmissions.CountGroupByOptionID(reqCtx, v.ID)
if err != nil {
return nil, errors.Wrap(err, "count options")
}
for _, o := range options {
count := counts[o.ID]
percent := 0
if total > 0 {
percent = int(count * 100 / total)
}
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
}
case db.VoteTypeMultiple:
options, _ := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
counts := make(map[uint]int64, len(options))
submissions, _ := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
var totalSelections int64
for _, s := range submissions {
var data struct {
Options []uint `json:"options"`
}
_ = json.Unmarshal(s.FormData, &data)
seen := make(map[uint]struct{}, len(data.Options))
for _, id := range data.Options {
if _, exists := seen[id]; exists {
continue
}
seen[id] = struct{}{}
counts[id]++
totalSelections++
}
}
for _, o := range options {
count := counts[o.ID]
percent := 0
// 多选:百分比分母取"总勾选数"(各项相加≈100%,符合"票数分布"直觉);
// res.Total 仍是参与人数,用于"总参与"展示。
if totalSelections > 0 {
percent = int(count * 100 / totalSelections)
}
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
}
case db.VoteTypeRating:
scoreCounts := make(map[int]int64)
var sum int64
submissions, _ := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
for _, s := range submissions {
var data struct {
Rating int `json:"rating"`
}
_ = json.Unmarshal(s.FormData, &data)
if data.Rating >= 1 && data.Rating <= 10 {
scoreCounts[data.Rating]++
sum += int64(data.Rating)
}
}
for score := 1; score <= 10; score++ {
res.Ratings = append(res.Ratings, &RatingStat{Score: score, Count: scoreCounts[score]})
}
if total > 0 {
res.AvgRating = float64(sum) / float64(total)
}
case db.VoteTypeRank:
options, _ := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
rankSums := make(map[uint]float64, len(options))
rankCounts := make(map[uint]int64, len(options))
submissions, _ := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
for _, s := range submissions {
var data struct {
Ranking map[string]float64 `json:"ranking"`
}
_ = json.Unmarshal(s.FormData, &data)
for idStr, r := range data.Ranking {
id, _ := strconv.ParseUint(idStr, 10, 64)
rankSums[uint(id)] += r
rankCounts[uint(id)]++
}
}
for _, o := range options {
var avg float64
if rankCounts[o.ID] > 0 {
avg = rankSums[o.ID] / float64(rankCounts[o.ID])
}
res.Rankings = append(res.Rankings, &RankStat{Option: o, AvgRank: avg})
}
var rankSum float64
for _, r := range res.Rankings {
rankSum += r.AvgRank
}
if len(res.Rankings) > 0 {
res.AvgRank = rankSum / float64(len(res.Rankings))
}
case db.VoteTypeForm:
res.FormFields, _ = db.VoteFormFields.ListByVoteID(reqCtx, v.ID)
res.FormCount = total
res.FormFieldStats = aggregateFormFieldStats(reqCtx, res.FormFields)
}
// 自定义补充选项聚合(single/multiple/rank 且开启 AllowCustom;评分/表单不适用)
if v.AllowCustom && (v.Type == db.VoteTypeSingle || v.Type == db.VoteTypeMultiple || v.Type == db.VoteTypeRank) {
res.CustomStats = aggregateCustomStats(reqCtx, v)
}
return res, nil
}
+17 -135
View File
@@ -72,6 +72,16 @@ func InvalidateResultsCache(voteID uint) {
resultsCache.Delete(strconv.FormatUint(uint64(voteID), 10))
}
// ExpireResultsCache 把某投票的结果缓存标记为过期,但保留旧值。
//
// 投票提交用这个,不用 InvalidateResultsCache:后者会把缓存删光,
// 于是每一票都让下一个读请求同步重算一次全量聚合(O(N)),
// 持续有人投票时就是 O(N²)。标记过期后,下一个读请求先拿到旧值,
// 再由后台按投票去重地重算。
func ExpireResultsCache(voteID uint) {
resultsCache.Expire(strconv.FormatUint(uint64(voteID), 10))
}
// censorCheck 对开放文本做内容审核。
//
// 与提问 / 私聊共用同一个守卫(internal/security/moderation.Check):是否拦截由
@@ -252,147 +262,17 @@ type Results struct {
FormFieldStats []*FormFieldStat // form:各字段聚合统计(混合问卷)
}
// LoadResults 按投票类型聚合统计(参与端结果页与管理后台结果页共用)。
func LoadResults(ctx context.Context, v *db.Vote) (*Results, error) {
// 短 TTL 缓存:避免每次请求都逐条 json.Unmarshal 全部提交记录(O(N) 重算)。
cacheKey := strconv.FormatUint(uint64(v.ID), 10)
if cached, ok := resultsCache.Get(cacheKey); ok {
return cached.(*Results), nil
}
res := &Results{Type: v.Type}
total, err := db.VoteSubmissions.CountByVoteID(ctx.Request().Context(), v.ID)
if err != nil {
return nil, errors.Wrap(err, "count submissions")
}
res.Total = total
switch v.Type {
case db.VoteTypeSingle:
options, _ := db.VoteOptions.ListByVoteID(ctx.Request().Context(), v.ID)
// 一次分组查询替代逐选项 COUNT(避免 N+1)
counts, err := db.VoteSubmissions.CountGroupByOptionID(ctx.Request().Context(), v.ID)
if err != nil {
return nil, errors.Wrap(err, "count options")
}
for _, o := range options {
count := counts[o.ID]
percent := 0
if total > 0 {
percent = int(count * 100 / total)
}
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
}
case db.VoteTypeMultiple:
options, _ := db.VoteOptions.ListByVoteID(ctx.Request().Context(), v.ID)
counts := make(map[uint]int64, len(options))
submissions, _ := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), v.ID, nil)
var totalSelections int64
for _, s := range submissions {
var data struct {
Options []uint `json:"options"`
}
_ = json.Unmarshal(s.FormData, &data)
seen := make(map[uint]struct{}, len(data.Options))
for _, id := range data.Options {
if _, exists := seen[id]; exists {
continue
}
seen[id] = struct{}{}
counts[id]++
totalSelections++
}
}
for _, o := range options {
count := counts[o.ID]
percent := 0
// 多选:百分比分母取"总勾选数"(各项相加≈100%,符合"票数分布"直觉);
// res.Total 仍是参与人数,用于"总参与"展示。
if totalSelections > 0 {
percent = int(count * 100 / totalSelections)
}
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
}
case db.VoteTypeRating:
scoreCounts := make(map[int]int64)
var sum int64
submissions, _ := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), v.ID, nil)
for _, s := range submissions {
var data struct {
Rating int `json:"rating"`
}
_ = json.Unmarshal(s.FormData, &data)
if data.Rating >= 1 && data.Rating <= 10 {
scoreCounts[data.Rating]++
sum += int64(data.Rating)
}
}
for score := 1; score <= 10; score++ {
res.Ratings = append(res.Ratings, &RatingStat{Score: score, Count: scoreCounts[score]})
}
if total > 0 {
res.AvgRating = float64(sum) / float64(total)
}
case db.VoteTypeRank:
options, _ := db.VoteOptions.ListByVoteID(ctx.Request().Context(), v.ID)
rankSums := make(map[uint]float64, len(options))
rankCounts := make(map[uint]int64, len(options))
submissions, _ := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), v.ID, nil)
for _, s := range submissions {
var data struct {
Ranking map[string]float64 `json:"ranking"`
}
_ = json.Unmarshal(s.FormData, &data)
for idStr, r := range data.Ranking {
id, _ := strconv.ParseUint(idStr, 10, 64)
rankSums[uint(id)] += r
rankCounts[uint(id)]++
}
}
for _, o := range options {
var avg float64
if rankCounts[o.ID] > 0 {
avg = rankSums[o.ID] / float64(rankCounts[o.ID])
}
res.Rankings = append(res.Rankings, &RankStat{Option: o, AvgRank: avg})
}
var rankSum float64
for _, r := range res.Rankings {
rankSum += r.AvgRank
}
if len(res.Rankings) > 0 {
res.AvgRank = rankSum / float64(len(res.Rankings))
}
case db.VoteTypeForm:
res.FormFields, _ = db.VoteFormFields.ListByVoteID(ctx.Request().Context(), v.ID)
res.FormCount = total
res.FormFieldStats = aggregateFormFieldStats(ctx, res.FormFields)
}
// 自定义补充选项聚合(single/multiple/rank 且开启 AllowCustom;评分/表单不适用)
if v.AllowCustom && (v.Type == db.VoteTypeSingle || v.Type == db.VoteTypeMultiple || v.Type == db.VoteTypeRank) {
res.CustomStats = aggregateCustomStats(ctx, v)
}
resultsCache.Set(cacheKey, res, resultsCacheTTL)
return res, nil
}
// aggregateFormFieldStats 统计表单投票(混合问卷)各字段的聚合结果:
// radio/select/checkbox → 选项票数;rating → 分布 + 平均分;rank → 各选项被排次数 + 平均排名;
// number → 平均值;text/textarea → 填写数(内容不聚合展示,保护隐私)。
func aggregateFormFieldStats(ctx context.Context, fields []*db.VoteFormField) []*FormFieldStat {
func aggregateFormFieldStats(reqCtx stdctx.Context, fields []*db.VoteFormField) []*FormFieldStat {
if len(fields) == 0 {
return nil
}
// 必须加载全部提交记录:radio/select/checkbox/rating/rank/number 等字段的统计
// 同样依赖 submissions,不能只在存在 text/textarea 字段时才加载。
// 否则无文本字段的表单投票结果页会全部显示为 0(缺陷)。
submissions, _ := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), fields[0].VoteID, nil)
submissions, _ := db.VoteSubmissions.ListByVoteID(reqCtx, fields[0].VoteID, nil)
return aggregateFormFieldStatsWithSubs(fields, submissions)
}
@@ -566,8 +446,8 @@ func addCustomStat(stats []CustomStat, text string) []CustomStat {
}
// aggregateCustomStats 统计自定义补充选项(按文本去重计数,保持首现顺序)。
func aggregateCustomStats(ctx context.Context, v *db.Vote) []CustomStat {
submissions, err := db.VoteSubmissions.ListByVoteID(ctx.Request().Context(), v.ID, nil)
func aggregateCustomStats(reqCtx stdctx.Context, v *db.Vote) []CustomStat {
submissions, err := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
if err != nil {
return nil
}
@@ -844,7 +724,9 @@ func Submit(ctx context.Context, recaptcha recaptcha.RecaptchaV3) {
// 写穿式失效:新提交成功后立即清除该投票的结果缓存,
// 避免结果页在 TTL 内显示陈旧计数,并把 O(N) 重算成本收敛到"每次新投票一次"。
InvalidateResultsCache(vote.ID)
// 标记过期而不删除:下一个读请求会先拿到旧值,再由后台重算。
// 用 InvalidateResultsCache 会让每一票都变成下一个读请求的同步全量重算(O(N²))。
ExpireResultsCache(vote.ID)
// 定量收集:箱主级累计一份;单份投票设置上限后达到即自动结束该投票。
if err := db.FeatureControls.IncrementQuota(ctx.Request().Context(), vote.UserID, db.FeatureVote, 1); err != nil {