perf: 流式错误判定预过滤、再减一个写索引，加可选明细保留期

- 流式放行前的上游错误判定加廉价子串预过滤（"error" / response.failed）：
  长思考阶段几十上百条事件不再逐条 json.Unmarshal。
- usage_logs 去掉 idx_channel_created：没有任何查询按 channel_id 过滤该表
  （渠道统计走 created_at 时间窗后在内存分组），复合索引只是每行一次多余的
  B-tree 插入；schema v9 迁移里显式 DROP 旧库索引。至此该表只剩
  (user_id, created_at)、created_at 两个真被用到的索引。
- 新增可选保留期（OT_USAGE_RETENTION_DAYS，默认 0=永久保留）：每天分批
  （2000 行/批、批间 50ms）清理过期 usage_logs，避免长期运行后时间窗查询
  与备份随表增长变慢；余额流水 balance_logs 不在清理范围。

参数复测（记账稳态吞吐，3 次取中位）：
- batch 64（当前）30.2k rows/s；batch 128 19.0k、batch 256 14.3k（更差，保持 64）
- wal_autocheckpoint(4000)、cache_size(-64MB) 与当前配置无差别，不引入

验证：go vet、go test ./...、-race（usage/proxy/store）全绿；新增保留期三项测试
（只删明细不动流水、未开启不动数据、跨分批删完）。

.env.example

@@ -64,6 +64,10 @@ OT_PROXY_HEALTH_FAIL_THRESHOLD=2
 # 原始体会截断：请求体 256 KiB、响应体 1 MiB，避免单条日志几十 MB 拖慢查询与备份。
 OT_PROXY_LOG_RAW=false
 
+# 用量明细保留天数（0=永久保留，默认）。开启后每天清理超过该天数的 usage_logs，
+# 分批删除（每批 2000 行），余额流水 balance_logs 不受影响。
+OT_USAGE_RETENTION_DAYS=0
+
 # 限流（内存计数，Redis 后置）：用户级每秒请求数上限（0=不限制）
 OT_RATELIMIT_USER_RPS=20
 # 认证端点限速（每分钟，0=不限制）：登录是密码爆破面，注册是批量建号面

README.md

@@ -17,6 +17,7 @@
 - **渠道体系**：按模型绑定选渠道 + 加权负载均衡；模型已有绑定时只用绑定渠道（都不可用则 503，不会把请求发给不认识它的渠道）；每渠道并发信号量（满载排队 `OT_PROXY_CHANNEL_ACQUIRE_WAIT`，默认 150ms，仍满则 503 + Retry-After 而不是 502）；后台健康检查（探测 `/models`，连续失败进 cooldown、恢复放回）+ 运行期健康（按失败原因分类冷却：网络错误/5xx 连续 3 次进冷却且指数退避，余额/配额类第一次即冷却 30 分钟，成功清零退避计数）；可安全重试的失败自动故障转移（网络错误/429/5xx/超时/流式首个内容事件之前的任何失败）；**会话亲和**（`OT_PROXY_AFFINITY_TTL`，默认 5m）：同一会话粘住上次应答的渠道，命中上游提示词缓存而不是全价重算，冷却/下线自动回退加权随机
 - **上游超时**：渠道 `timeout_ms` 生效——非流式为整段请求上限，流式仅约束首包（正文不限时长）；上游请求绑定客户端 context，断连即取消
 - **故障转移与流式细节**（借鉴 magpie 的 holdWriter）：流式请求在**首个内容事件**到达前由网关缓冲——上游的错误事件、空流、断连、协议形态不符（要流给了整包，didNotStream）都在放行前换渠道消化，客户端无感知；`OT_PROXY_FIRST_TOKEN_WAIT` 限定放行前等内容的最长时间（0=不限）；429 不一定是限流——错误体带余额/配额特征时按配额类长冷却；上游 `Retry-After` 会延长该渠道冷却（夹在 10 分钟内）；重试受 `OT_PROXY_FAILOVER_BUDGET` 时间预算约束（首个渠道不受限，避免坏渠道把客户端拖到 N×timeout）；流式空闲超过 `OT_PROXY_STREAM_HEARTBEAT` 补一行 SSE 注释心跳（扣住期间也照发），防止 nginx/Cloudflare 的空闲超时掐掉长思考；响应头已被心跳先行写出时，全部候选失败以 SSE 错误事件收尾
+- **数据保留**：`OT_USAGE_RETENTION_DAYS`（默认 0=永久保留）开启后每天分批清理过期用量明细；余额流水不清理
 - **渠道观测**：`GET /api/v1/admin/channels/stats?window=5m|1h|6h|24h|7d` 按渠道聚合请求量、成功率、错误码分布与 P50/P95 延迟；渠道页直接展示近况与运行期冷却剩余时间，可一键恢复
 - **限流/配额**（内存计数）：密钥级每日请求数 / Token 数配额、用户级每秒速率（`OT_RATELIMIT_USER_RPS`）；超限返回 429
 - **用量计费**：请求级 `usage_logs` 异步批量落库，按模型价格扣减余额，日粒度预聚合（`usage_daily`）；模型可单独配置缓存写价（留空按输入价 ×1.25）；记录思考 token（`reasoning_tokens`，展示口径不重复计费）、Anthropic 1h 缓存写（计价 2× 输入）与上游实际应答的模型名（`served_model`，映射/聚合商场景对账用）

server/internal/app/app.go

@@ -19,18 +19,20 @@ import (
 )
 
 type App struct {
-	Cfg       *config.Config
-	DB        *gorm.DB
-	Hasher    *crypto.PasswordHasher
-	Enc       *crypto.Encryptor
-	JWT       *jwt.Manager
-	Usage     *usage.Recorder
-	Health    *channel.HealthMonitor
-	Limit     *ratelimit.Limiter
-	Passkeys  *passkey.Service
-	startedAt time.Time
-	ctx       context.Context
-	cancel    context.CancelFunc
+	Cfg      *config.Config
+	DB       *gorm.DB
+	Hasher   *crypto.PasswordHasher
+	Enc      *crypto.Encryptor
+	JWT      *jwt.Manager
+	Usage    *usage.Recorder
+	Health   *channel.HealthMonitor
+	Limit    *ratelimit.Limiter
+	Passkeys *passkey.Service
+	// usageRetention 用量明细保留期任务（未开启时为 no-op）
+	usageRetention func()
+	startedAt      time.Time
+	ctx            context.Context
+	cancel         context.CancelFunc
 }
 
 func New(cfg *config.Config) (*App, error) {
@@ -61,6 +63,9 @@ func New(cfg *config.Config) (*App, error) {
 	})
 	a.Health.Start(a.ctx)
 
+	// 用量明细保留期（默认 0=永久保留，只清明细不动余额流水）
+	a.usageRetention = usage.StartRetention(a.ctx, db, cfg.Usage.RetentionDays)
+
 	a.Passkeys, err = passkey.New(db, passkey.Config{
 		RPID: cfg.WebAuthn.RPID, Origin: cfg.WebAuthn.RPOrigin, Name: cfg.WebAuthn.RPName,
 	})
@@ -71,6 +76,9 @@ func New(cfg *config.Config) (*App, error) {
 }
 
 func (a *App) Close() {
+	if a.usageRetention != nil {
+		a.usageRetention()
+	}
 	a.Usage.Close()
 }
 

server/internal/config/config.go

@@ -18,6 +18,7 @@ type Config struct {
 	JWT       JWTConfig
 	Auth      AuthConfig
 	Proxy     ProxyConfig
+	Usage     UsageConfig
 	RateLimit RateLimitConfig
 	WebAuthn  WebAuthnConfig
 	Master    string // 渠道密钥 AES-GCM 主密钥（来自环境变量）
@@ -32,6 +33,13 @@ type Config struct {
 	TrustedProxies []string
 }
 
+// UsageConfig 用量数据的保留策略。
+type UsageConfig struct {
+	// RetentionDays 用量明细保留天数；0（默认）= 永久保留。
+	// 只清 usage_logs，不动余额流水。
+	RetentionDays int
+}
+
 // WebAuthnConfig Passkey（WebAuthn）配置。
 type WebAuthnConfig struct {
 	RPID     string // Relying Party ID（域名，如 localhost）
@@ -156,6 +164,7 @@ func Load() (*Config, error) {
 	v.SetDefault("proxy.affinity_ttl", "5m")
 	v.SetDefault("proxy.channel_acquire_wait", "150ms")
 	v.SetDefault("proxy.auth_cache_ttl", "2s")
+	v.SetDefault("usage.retention_days", 0)
 	v.SetDefault("proxy.health_interval", "60s")
 	v.SetDefault("proxy.health_fail_threshold", 2)
 	v.SetDefault("proxy.log_raw", false)
@@ -213,6 +222,9 @@ func Load() (*Config, error) {
 			HealthFailThreshold: v.GetInt("proxy.health_fail_threshold"),
 			LogRaw:              v.GetBool("proxy.log_raw"),
 		},
+		Usage: UsageConfig{
+			RetentionDays: v.GetInt("usage.retention_days"),
+		},
 		RateLimit: RateLimitConfig{
 			UserRPS:          v.GetInt("ratelimit.user_rps"),
 			LoginIPPerMin:    v.GetInt("ratelimit.login_ip_per_min"),

server/internal/proxy/hold.go

@@ -242,6 +242,11 @@ func sseEventHasContent(line []byte, proto string) bool {
 // 返回错误消息文本；不是错误事件时 ok=false。只在放行前判定——放行后错误
 // 属于客户端可见的回复内容，不能拿去换渠道。
 func sseUpstreamError(line []byte) (msg string, ok bool) {
+	// 廉价的子串预过滤：绝大多数事件与错误无关，先排除再谈解析。
+	// 放行前每条事件都会过这里，长思考阶段的几十上百条事件不值得逐条反序列化。
+	if !bytes.Contains(line, []byte(`"error"`)) && !bytes.Contains(line, []byte("response.failed")) {
+		return "", false
+	}
 	s, ok := sseDataPayload(line)
 	if !ok {
 		return "", false

server/internal/store/db.go

@@ -27,7 +27,8 @@ import (
 // idx_user_created 是 user_id 单列，光改 struct tag 不会重建，需删掉重迁一次。
 // v7：usage_logs 加 reasoning_tokens / cache_write_1h_tokens / served_model 三列。
 // v8：去掉 usage_logs 上没人查询的 key_id / trace_id 索引（每行两次多余的 B-tree 插入）。
-const schemaVersion = 8
+// v9：同样去掉没人查询的 idx_channel_created。
+const schemaVersion = 9
 
 // Open 打开数据库连接并自动迁移。
 // 开发默认 SQLite（自动追加 WAL 与 busy_timeout，见 withSQLitePragmas），生产可切 postgres。
@@ -137,7 +138,7 @@ func withSQLitePragmas(dsn string) string {
 // dropUnusedUsageLogIndexes 删除历史库里已经没用的索引：AutoMigrate 不会自动删索引，
 // 去掉 struct tag 只是不再新建，旧库上它们还会继续拖慢每次插入。
 func dropUnusedUsageLogIndexes(db *gorm.DB) {
-	for _, name := range []string{"idx_usage_logs_key_id", "idx_usage_logs_trace_id"} {
+	for _, name := range []string{"idx_usage_logs_key_id", "idx_usage_logs_trace_id", "idx_channel_created"} {
 		if err := db.Exec("DROP INDEX IF EXISTS " + name).Error; err != nil {
 			log.Printf("store: drop index %s: %v", name, err)
 		}

server/internal/store/models.go

@@ -187,10 +187,12 @@ type UsageLog struct {
 	RequestID string `gorm:"size:128" json:"request_id"` // 上游 request id
 	// TraceID/KeyID 不建索引：没有查询按它们过滤（幂等检查走 balance_logs.ref_id），
 	// 每个索引都是每行一次 B-tree 插入，在记账这条最热的写路径上纯属写放大。
-	TraceID             string  `gorm:"size:64" json:"trace_id"`
-	UserID              uint64  `gorm:"index:idx_user_created,priority:1;not null" json:"user_id"`
-	KeyID               uint64  `json:"key_id"`
-	ChannelID           uint64  `gorm:"index:idx_channel_created,priority:1" json:"channel_id"`
+	TraceID string `gorm:"size:64" json:"trace_id"`
+	UserID  uint64 `gorm:"index:idx_user_created,priority:1;not null" json:"user_id"`
+	KeyID   uint64 `json:"key_id"`
+	// 与 key_id/trace_id 同理：没有任何查询按 channel_id 过滤 usage_logs
+	// （渠道统计走 created_at 时间窗后在内存里分组），复合索引纯属写放大。
+	ChannelID           uint64  `json:"channel_id"`
 	ModelID             uint64  `json:"model_id"`
 	ModelName           string  `gorm:"size:128" json:"model_name"`
 	Protocol            string  `gorm:"size:32" json:"protocol"` // responses|chat|messages
@@ -210,8 +212,9 @@ type UsageLog struct {
 	ErrorCode           *string `json:"error_code,omitempty"`
 	RawRequest          string  `gorm:"type:text" json:"raw_request"`  // 客户端原始请求体（未转换）
 	RawResponse         string  `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件）
-	// CreatedAt 同时参与 user/channel 两个复合索引（时间窗过滤），自身单列索引保留给全表时间扫
-	CreatedAt time.Time `gorm:"index;index:idx_user_created,priority:2;index:idx_channel_created,priority:2" json:"created_at"`
+	// CreatedAt 参与 (user, created) 复合索引（用户侧时间窗过滤），单列索引服务
+	// 全表时间窗扫描（渠道统计、运营总览）。
+	CreatedAt time.Time `gorm:"index;index:idx_user_created,priority:2" json:"created_at"`
 }
 
 // UsageDaily 日粒度预聚合（PLANNING §6.6）

server/internal/usage/retention.go

@@ -0,0 +1,86 @@
+package usage
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"gorm.io/gorm"
+)
+
+// 记账明细的保留期清理。
+//
+// usage_logs 只增不减：长期运行的实例上，渠道统计这类「按时间窗扫描」的查询会
+// 随表增长线性变慢，备份也会越来越大。余额流水（balance_logs）不在清理范围内——
+// 那是财务记录，需要长期保留。
+//
+// 默认关闭（retention_days=0）：删数据是不可逆操作，由部署者显式开启。
+// 清理分批进行（每批 2000 行、批间小睡），避免长时间持锁影响在线请求。
+
+const (
+	retentionBatch = 2000
+	retentionTick  = 24 * time.Hour
+)
+
+// StartRetention 周期性删除超过 days 天的用量明细；days<=0 时不启动。
+// 返回停止函数（进程退出时调用）。
+func StartRetention(ctx context.Context, db *gorm.DB, days int) (stop func()) {
+	if days <= 0 || db == nil {
+		return func() {}
+	}
+	done := make(chan struct{})
+	go func() {
+		purgeOlderThan(ctx, db, days)
+		t := time.NewTicker(retentionTick)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-done:
+				return
+			case <-t.C:
+				purgeOlderThan(ctx, db, days)
+			}
+		}
+	}()
+	return func() {
+		select {
+		case <-done:
+		default:
+			close(done)
+		}
+	}
+}
+
+// purgeOlderThan 分批删除 cutoff 之前的明细，返回删除行数。
+func purgeOlderThan(ctx context.Context, db *gorm.DB, days int) int64 {
+	cutoff := time.Now().UTC().AddDate(0, 0, -days)
+	var total int64
+	for {
+		if ctx.Err() != nil {
+			return total
+		}
+		res := db.Exec(
+			"DELETE FROM usage_logs WHERE id IN (SELECT id FROM usage_logs WHERE created_at < ? LIMIT ?)",
+			cutoff, retentionBatch)
+		if res.Error != nil {
+			log.Printf("usage: retention purge failed: %v", res.Error)
+			return total
+		}
+		n := res.RowsAffected
+		total += n
+		if n < retentionBatch {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return total
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+	if total > 0 {
+		log.Printf("usage: retention removed %d rows older than %s", total, cutoff.Format("2006-01-02"))
+	}
+	return total
+}

server/internal/usage/retention_test.go

@@ -0,0 +1,84 @@
+package usage
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/openteam/server/internal/store"
+)
+
+// 保留期只删过期明细，余额流水一条不动；未开启（0）时什么都不做。
+func TestRetentionPurgesOldLogsOnly(t *testing.T) {
+	db := openTestDB(t)
+	u := seedUser(t, db, 100)
+
+	old := time.Now().UTC().AddDate(0, 0, -40)
+	fresh := time.Now().UTC()
+	for i := 0; i < 5; i++ {
+		db.Create(&store.UsageLog{TraceID: fmt.Sprintf("old-%d", i), UserID: u, ModelName: "m1",
+			Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: old})
+	}
+	for i := 0; i < 3; i++ {
+		db.Create(&store.UsageLog{TraceID: fmt.Sprintf("new-%d", i), UserID: u, ModelName: "m1",
+			Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: fresh})
+	}
+	db.Create(&store.BalanceLog{UserID: u, Change: -0.05, BalanceAfter: 99.95,
+		Type: store.BalanceTypeUsage, RefID: "usage:old-0"})
+
+	n := purgeOlderThan(context.Background(), db, 30)
+	if n != 5 {
+		t.Fatalf("应删除 5 条过期明细，实际 %d", n)
+	}
+	var logs, ledgers int64
+	db.Model(&store.UsageLog{}).Count(&logs)
+	db.Model(&store.BalanceLog{}).Count(&ledgers)
+	if logs != 3 {
+		t.Fatalf("应保留 3 条新明细，实际 %d", logs)
+	}
+	if ledgers != 1 {
+		t.Fatalf("余额流水不应被清理，实际 %d 条", ledgers)
+	}
+}
+
+// 未配置保留期时不启动、不删数据。
+func TestRetentionDisabled(t *testing.T) {
+	db := openTestDB(t)
+	u := seedUser(t, db, 100)
+	db.Create(&store.UsageLog{TraceID: "x", UserID: u, ModelName: "m1",
+		Cost: 0.01, Status: store.UsageStatusSuccess,
+		CreatedAt: time.Now().UTC().AddDate(0, 0, -400)})
+
+	stop := StartRetention(context.Background(), db, 0)
+	stop()
+	var n int64
+	db.Model(&store.UsageLog{}).Count(&n)
+	if n != 1 {
+		t.Fatalf("未开启保留期时不应删除数据，实际剩 %d", n)
+	}
+}
+
+// 分批删除：行数超过一批时应全部删完。
+func TestRetentionBatches(t *testing.T) {
+	db := openTestDB(t)
+	u := seedUser(t, db, 100)
+	old := time.Now().UTC().AddDate(0, 0, -90)
+	logs := make([]store.UsageLog, 0, retentionBatch+50)
+	for i := 0; i < retentionBatch+50; i++ {
+		logs = append(logs, store.UsageLog{TraceID: fmt.Sprintf("b-%d", i), UserID: u,
+			ModelName: "m1", Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: old})
+	}
+	if err := db.CreateInBatches(logs, 500).Error; err != nil {
+		t.Fatal(err)
+	}
+	n := purgeOlderThan(context.Background(), db, 30)
+	if n != int64(len(logs)) {
+		t.Fatalf("应删除 %d 条，实际 %d", len(logs), n)
+	}
+	var left int64
+	db.Model(&store.UsageLog{}).Count(&left)
+	if left != 0 {
+		t.Fatalf("应全部删完，实际剩 %d", left)
+	}
+}