weichen
5cd67be982
ci / vet-and-test (pull_request) Successful in 41s
Why: the approved plan U3 (KTD3/KTD7/KTD10 plus minimal recovery isolation) requires concurrent submits to not silently overwrite, unknown-side-effect runs to never auto-resume, and the shared rate limiter to degrade conservatively instead of failing open. Evidence: full backend suite green twice-relevant paths against real PostgreSQL 18.6 and Redis (isolated instance) and on SQLite; new tests - card_version_test (concurrent single winner), recovery_hold_test + recovery_holds_test (skip/release/fail-closed CRUD), ratelimit atomic-window and degraded-fallback tests. Impact: AppendMessageWithRequest now allocates seq under the conversation advisory lock in one transaction; activity_cards gain version (v19) with SaveCardVersioned/GetCardVersion/CardVersion and cardsvc user mutations (status/pin/action/resolve-by-ref) return 409-mapped conflicts while system mirrors keep blind writes; recovery_holds (v20) with admin list/release endpoints and fail-closed resume skip; Redis limiter uses one Lua INCR+PEXPIRE script and falls back to a per-instance limit with a Degraded counter. Refs: proposals/2026-10-11-collaboration-storage-and-distribution.md U3 KTD3/KTD7/KTD10; R3/R6/R7/R8.
98 lines
2.8 KiB
Go
98 lines
2.8 KiB
Go
// Package ratelimit 提供按租户的限流（阶段 3）：
|
||
// - Memory：进程内固定窗口（默认，单实例档，零外部服务）；
|
||
// - Redis：多实例共享固定窗口（原生 Redis，设置 runtime.redisAddr 启用）。
|
||
package ratelimit
|
||
|
||
import (
|
||
"context"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
type Limiter interface {
|
||
// Allow 判定 key（租户 owner）在窗口内第 N 次请求是否放行。
|
||
Allow(ctx context.Context, key string, limit int, window time.Duration) bool
|
||
}
|
||
|
||
// ---- 进程内固定窗口 ----
|
||
|
||
type memWindow struct {
|
||
n int
|
||
reset time.Time
|
||
}
|
||
|
||
type Memory struct {
|
||
mu sync.Mutex
|
||
windows map[string]*memWindow
|
||
now func() time.Time
|
||
}
|
||
|
||
func NewMemory() *Memory {
|
||
return &Memory{windows: map[string]*memWindow{}, now: time.Now}
|
||
}
|
||
|
||
func (m *Memory) Allow(_ context.Context, key string, limit int, window time.Duration) bool {
|
||
if limit <= 0 {
|
||
return true
|
||
}
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
now := m.now()
|
||
w, ok := m.windows[key]
|
||
if !ok || !now.Before(w.reset) {
|
||
m.windows[key] = &memWindow{n: 1, reset: now.Add(window)}
|
||
return true
|
||
}
|
||
if w.n >= limit {
|
||
return false
|
||
}
|
||
w.n++
|
||
return true
|
||
}
|
||
|
||
// ---- Redis 固定窗口（Lua 原子 INCR+PEXPIRE；故障降级进程内保守限流）----
|
||
|
||
// incrExpire 原子计数并只在首击设窗（协作存储 U3，KTD10）：两步 INCR/EXPIRE
|
||
// 在 EXPIRE 失败或进程于其间退出时会留下永不过期的键；Lua 单脚本消除该窗口。
|
||
var incrExpire = redis.NewScript(`
|
||
local n = redis.call('INCR', KEYS[1])
|
||
if n == 1 then redis.call('PEXPIRE', KEYS[1], ARGV[1]) end
|
||
return n
|
||
`)
|
||
|
||
type Redis struct {
|
||
rdb *redis.Client
|
||
// fallback：Redis 故障时的进程内保守限流（同限值、按实例计数——是共享
|
||
// 限流的近似而非替代；昂贵入口的硬保护在 PG admission，不依赖本层）。
|
||
fallback *Memory
|
||
degraded atomic.Int64
|
||
}
|
||
|
||
func NewRedis(addr, password string, db int) *Redis {
|
||
return &Redis{rdb: redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db}), fallback: NewMemory()}
|
||
}
|
||
|
||
func (r *Redis) Allow(ctx context.Context, key string, limit int, window time.Duration) bool {
|
||
if limit <= 0 {
|
||
return true
|
||
}
|
||
rkey := "rl:" + key + ":" + time.Now().UTC().Truncate(window).Format("150405")
|
||
n, err := incrExpire.Run(ctx, r.rdb, []string{rkey}, window.Milliseconds()).Int64()
|
||
if err != nil {
|
||
// 降级（KTD10）：不再无界放行；按实例保守限流并计数供观测告警。
|
||
r.degraded.Add(1)
|
||
return r.fallback.Allow(ctx, key, limit, window)
|
||
}
|
||
return n <= int64(limit)
|
||
}
|
||
|
||
// Degraded 返回累计降级次数（观测：非零说明 Redis 不可用，需运维介入）。
|
||
func (r *Redis) Degraded() int64 { return r.degraded.Load() }
|
||
|
||
func (r *Redis) Ping(ctx context.Context) error { return r.rdb.Ping(ctx).Err() }
|
||
|
||
func (r *Redis) Close() error { return r.rdb.Close() }
|