superidou
9501751792
三轮审查修复(60+ 项),相对远端 main(b536672)的关键变更:
- 安全: 数据面 SSRF 拨号防护(防 DNS rebinding)/上游凭据剥离/登录防枚举
与锁定态统一/可信代理(X-Forwarded-For)限流加固/会话版本失效机制/
撤销即时传播/弱密钥拒绝启动/脱敏字节级重写(保签名契约)
- 业务逻辑: 裸 body 上传 panic/bootstrap 审计管线卡死/定价通配符优先级/
全局工具可见性/调度器停机补跑/TOTP 挑战令牌消费顺序/熔断探针语义/
>4MB 响应 token 计量/管理员重置密码作废会话 等
- 前端: 新 logo(语枢 AI 网关主题)/Provider 凭据异常警示/删除入口/
后端错误消息透传/localStorage 敏感数据收敛
- 部署: CREDENTIAL_MASTER_KEY 持久化与弱值拒绝/Provider DELETE 接口/
nginx 安全头/worker 内存限制
- 新增迁移 000029(key_hash 索引)/000030(usage_daily 币种维度)
97 lines
2.6 KiB
Go
97 lines
2.6 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"math"
|
|
"time"
|
|
)
|
|
|
|
type WorkerConfig struct {
|
|
WorkerID string
|
|
BatchSize int
|
|
PollInterval time.Duration
|
|
Lease time.Duration
|
|
MaxAttempts int
|
|
MaxBackoff time.Duration
|
|
}
|
|
|
|
type Worker struct {
|
|
store *Store
|
|
publisher *RedisPublisher
|
|
config WorkerConfig
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func NewWorker(store *Store, publisher *RedisPublisher, config WorkerConfig, logger *slog.Logger) *Worker {
|
|
return &Worker{store: store, publisher: publisher, config: config, logger: logger}
|
|
}
|
|
|
|
func (w *Worker) Run(ctx context.Context) error {
|
|
for {
|
|
processed, err := w.runBatch(ctx)
|
|
if err != nil && ctx.Err() == nil && w.logger != nil {
|
|
w.logger.Error("outbox batch failed", "error", err)
|
|
}
|
|
if ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
if err == nil && processed > 0 {
|
|
continue
|
|
}
|
|
timer := time.NewTimer(w.config.PollInterval)
|
|
select {
|
|
case <-ctx.Done():
|
|
timer.Stop()
|
|
return nil
|
|
case <-timer.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *Worker) runBatch(ctx context.Context) (int, error) {
|
|
// Gate on the downstream publisher before claiming anything. Claiming
|
|
// increments each event's attempt counter, so claiming during a Redis
|
|
// outage would burn every queued event's delivery budget and dead-letter
|
|
// the whole queue the moment the budget ran out — even though the events
|
|
// themselves were never at fault. When Redis is unreachable we back off
|
|
// instead, leaving events untouched in PostgreSQL until it recovers.
|
|
if err := w.publisher.Ping(ctx); err != nil {
|
|
return 0, err
|
|
}
|
|
events, err := w.store.Claim(ctx, w.config.WorkerID, w.config.BatchSize, w.config.Lease)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
var batchErr error
|
|
for _, event := range events {
|
|
result, publishErr := w.publisher.Publish(ctx, event)
|
|
if publishErr == nil {
|
|
if err := w.store.MarkProcessed(ctx, event.EventID, w.config.WorkerID, result.StreamID); err != nil {
|
|
batchErr = errors.Join(batchErr, err)
|
|
}
|
|
continue
|
|
}
|
|
delay := retryDelay(event.Attempts, w.config.MaxBackoff)
|
|
if err := w.store.MarkFailed(ctx, event, w.config.WorkerID, publishErr, w.config.MaxAttempts, delay); err != nil {
|
|
batchErr = errors.Join(batchErr, err)
|
|
}
|
|
}
|
|
return len(events), batchErr
|
|
}
|
|
|
|
func retryDelay(attempt int, maximum time.Duration) time.Duration {
|
|
// attempt ≥ 35 时 2^(attempt-1) 秒会溢出 int64 纳秒,得到负 duration,
|
|
// 使 MarkFailed 把 available_at 设到过去,事件立即被重新认领形成热循环。
|
|
if attempt > 30 {
|
|
return maximum
|
|
}
|
|
seconds := math.Pow(2, float64(max(attempt-1, 0)))
|
|
delay := time.Duration(seconds * float64(time.Second))
|
|
if delay > maximum {
|
|
return maximum
|
|
}
|
|
return delay
|
|
}
|