diff --git a/.trellis/spec/backend/database-guidelines.md b/.trellis/spec/backend/database-guidelines.md index 95aa7f6..87b7808 100644 --- a/.trellis/spec/backend/database-guidelines.md +++ b/.trellis/spec/backend/database-guidelines.md @@ -24,7 +24,20 @@ Questions to answer: -(To be filled by the team) +### 租户级数据删除 + +- 租户主体与本地关联数据必须在同一 GORM transaction 中按“子记录→父记录”删除,每步错误用 `%w` 返回,不得忽略。 +- JSON payload/Setting 键等非外键引用要显式枚举并改写;不能只依赖 `AutoMigrate` 或 ORM association 推断级联范围。 +- 与后台任务、Webhook、解析器等并发写入交叉时,先定义全局一致的行锁顺序,并在写入前重新确认父记录存在。 +- 进程内 cron/缓存只在事务提交后同步;客户端取消不应中断已提交删除的必需运行时对齐。 +- 批量删除/清理遇到**无法解析的 JSON payload** 时记 `log.Printf` 警告并跳过该行(原样保留),不 fail-closed 阻断整个流程——坏数据不应把删除逼到手工修库(tenantdelete.go `logSkippedTask`)。 + +### 大集合谓词用子查询,禁止展开 IN 列表 + +- 行数无上界的集合(日志事件、调用日志等)做关联删除/查询时,一律 `WHERE x IN (SELECT …)` 子查询,**不要**先 Pluck ID 再 `IN ?` 展开:绑定变量有硬上限(modernc SQLite 32766,MySQL/PG 65535),数万行即失败且重试无解。 +- GORM 写法:把 `tx.Model(&T{}).Select("id").Where(...)` 作为参数传入 `Where("x IN (?)", sub)`(见 tenantdelete.go `deleteAlertHits` / `alertHitRuleIDs`)。 +- 若原本的 Pluck 兼有 `FOR UPDATE` 锁定语义,保留锁定 SELECT 本身,只是不再把结果拼进后续 SQL(`lockTenantEventRows`)。 +- 行数有小上界的集合(渠道、规则等配置类)可以继续用内存 ID 列表。 --- @@ -48,6 +61,14 @@ Questions to answer: +### Common Mistake: 用 `Save` 持久化在途任务的陈旧快照 + +**Symptom**:一条记录已被另一事务删除,在途任务随后执行 `Save` 却把该行重新插入,或覆盖并发更新后的 payload。 + +**Cause**:GORM `Save` 在 `UPDATE` 零命中时会回退到 `CREATE`/upsert,不适合持久化长时运行开始时读取的快照。 + +**Fix**:用 `WHERE id = ? AND updated_at = ?` 的条件 `Updates`,并严格要求 `RowsAffected == 1`;零命中表示记录已删除或版本已变,不得补做 `Create`。 + ### Common Mistake: gorm 读 NULL 列到已有值的结构体字段不会清零 **Symptom**:UPDATE 把可空列(如 `*time.Time`)写成 NULL 后,用**同一个结构体变量**再次 `First()` 读回,该字段仍是旧值;而新变量读取正常。断言/返回值出现「幽灵旧值」。 @@ -66,3 +87,19 @@ db.Model(&model.AiChannel{}). Where("id = ? AND (fail_count > 0 OR disabled_until IS NOT NULL)", id). Updates(map[string]any{"fail_count": 0, "disabled_until": gorm.Expr("NULL")}) ``` + +### `serializer:json` 字段走 map Updates 时手动 marshal + +- 切片/结构体字段用 `gorm:"serializer:json;type:text"` 声明(如 `AiKey.Models []string`),Create/First/struct 路径自动序列化; +- 但 `Updates(map[string]any{...})` 路径不要依赖 GORM 对 map 值应用 serializer——把值 `json.Marshal` 成 string 放进 map(见 aigateway.go `UpdateKey`),行为版本无关且可测; +- 存储格式与 serializer 一致(JSON 文本),读回仍走自动反序列化;`nil` 切片 marshal 为 `null`,读回 nil,天然表达「空 = 不限」语义。 + +### Common Mistake: 进程内缓存键漏掉查询维度 + +**Symptom**:多区间(compartment)租户在前端切换区间后,实例/卷/VCN 列表短暂显示上一个区间的数据(TTL 窗口内)。 + +**Cause**:`internal/oci/cached.go` 的 `ckey` 只拼了租户 OCID+资源名+region,而底层查询按 `cred.EffectiveCompartment()` 过滤——影响结果的维度没有全部进键,不同参数命中同一条缓存。 + +**Fix**:键值加入 `cred.CompartmentID`(空 = 租户根,天然区分)。 + +**Prevention**:缓存键必须覆盖影响回源结果的**全部**输入维度(租户、区间、区域、过滤参数);给 Credentials/查询结构体新增会改变结果的字段时,同步检查 `ckey` 调用点;隔离行为写进 `cached_test.go` 的 isolation 用例。 diff --git a/internal/oci/cached.go b/internal/oci/cached.go index 627b86d..fbeac89 100644 --- a/internal/oci/cached.go +++ b/internal/oci/cached.go @@ -32,8 +32,10 @@ func NewCachedClient(inner Client) *CachedClient { } // ckey 组缓存键;租户 OCID 在最前,写失效按前缀一锅端。 +// compartment 必须参与键值:列表查询按 EffectiveCompartment 过滤, +// 同租户切换区间时若共用键会串到上一个区间的缓存结果。 func ckey(cred Credentials, parts ...string) string { - return cred.TenancyOCID + "|" + strings.Join(parts, "|") + return cred.TenancyOCID + "|" + cred.CompartmentID + "|" + strings.Join(parts, "|") } // bust 写操作成功后失效该租户全部读缓存。 diff --git a/internal/oci/cached_test.go b/internal/oci/cached_test.go index 119e4bb..de14f28 100644 --- a/internal/oci/cached_test.go +++ b/internal/oci/cached_test.go @@ -49,6 +49,13 @@ func TestCachedClientHitAndIsolation(t *testing.T) { if inner.instCalls != 3 { t.Errorf("跨租户/区域回源 %d 次, want 3", inner.instCalls) } + // 同租户不同 compartment 各自回源,不得共用缓存 + inCompartment := testCred("t1") + inCompartment.CompartmentID = "ocid1.compartment.a" + _, _ = c.ListInstances(ctx, inCompartment, "r1") + if inner.instCalls != 4 { + t.Errorf("跨 compartment 回源 %d 次, want 4", inner.instCalls) + } } func TestCachedClientWriteBusts(t *testing.T) { diff --git a/internal/service/aigateway_chat.go b/internal/service/aigateway_chat.go index 8384815..5324ce0 100644 --- a/internal/service/aigateway_chat.go +++ b/internal/service/aigateway_chat.go @@ -27,27 +27,27 @@ type aiCandidate struct { modelOcid string } -// RespPassthrough 编排一次非流式直通调用:选渠道(priority→加权随机)→ 调用 → -// 可重试错误换渠道(整请求上限 3 次)并维护熔断;group 非空时只在同分组渠道内路由。 -// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。 -func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) { +// routeRetry 统一编排「选渠道(priority→加权随机)→ 调用 → 可重试错误换渠道」, +// 整请求上限 3 次并维护熔断;group 非空时只在同分组渠道内路由。 +func routeRetry[T any](ctx context.Context, s *AiGatewayService, modelName, group, capability string, once func(*aiCandidate) (T, error)) (T, ChatMeta, error) { + var zero T meta := ChatMeta{} excluded := map[uint]bool{} var lastErr error for attempt := 0; attempt < 3; attempt++ { - cand, err := s.pick(ctx, modelName, group, "CHAT", excluded) + cand, err := s.pick(ctx, modelName, group, capability, excluded) if err != nil { - return nil, meta, firstErr(lastErr, err) + return zero, meta, firstErr(lastErr, err) } meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name - payload, err := s.passthroughOnce(ctx, cand, raw) + out, err := once(cand) if err == nil { s.markSuccess(ctx, cand.ch.ID) - return payload, meta, nil + return out, meta, nil } retry, penalize := switchable(err) if !retry { - return nil, meta, err + return zero, meta, err } if penalize { s.markFailure(ctx, cand.ch.ID) @@ -56,7 +56,15 @@ func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, mode meta.Retries++ lastErr = err } - return nil, meta, lastErr + return zero, meta, lastErr +} + +// RespPassthrough 编排一次非流式直通调用。 +// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。 +func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) { + return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) ([]byte, error) { + return s.passthroughOnce(ctx, cand, raw) + }) } func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidate, raw []byte) ([]byte, error) { @@ -70,36 +78,13 @@ func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidat // RespPassthroughStream 编排流式直通:流建立成功即绑定渠道,建立失败按 switchable // 换渠道重试;建立后的中断不重试、不计熔断(与 OpenStream 语义一致)。 func (s *AiGatewayService) RespPassthroughStream(ctx context.Context, raw []byte, modelName, group string) (io.ReadCloser, ChatMeta, error) { - meta := ChatMeta{} - excluded := map[uint]bool{} - var lastErr error - for attempt := 0; attempt < 3; attempt++ { - cand, err := s.pick(ctx, modelName, group, "CHAT", excluded) - if err != nil { - return nil, meta, firstErr(lastErr, err) - } - meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name + return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) (io.ReadCloser, error) { cred, err := s.configs.credentialsByID(ctx, cand.ch.OciConfigID) if err != nil { - return nil, meta, err + return nil, err } - stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait()) - if err == nil { - s.markSuccess(ctx, cand.ch.ID) - return stream, meta, nil - } - retry, penalize := switchable(err) - if !retry { - return nil, meta, err - } - if penalize { - s.markFailure(ctx, cand.ch.ID) - } - excluded[cand.ch.ID] = true - meta.Retries++ - lastErr = err - } - return nil, meta, lastErr + return s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait()) + }) } // firstErr 在换渠道后仍失败时优先返回上游错误(而非「无渠道」)。 @@ -221,34 +206,11 @@ func weightedPick(chs []model.AiChannel) model.AiChannel { return chs[len(chs)-1] } -// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道(整请求上限 3 次)。 +// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道。 func (s *AiGatewayService) Embeddings(ctx context.Context, req aiwire.EmbeddingsRequest, group string) (*aiwire.EmbeddingsResponse, ChatMeta, error) { - meta := ChatMeta{} - excluded := map[uint]bool{} - var lastErr error - for attempt := 0; attempt < 3; attempt++ { - cand, err := s.pick(ctx, req.Model, group, "EMBEDDING", excluded) - if err != nil { - return nil, meta, firstErr(lastErr, err) - } - meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name - resp, err := s.embedOnce(ctx, cand, req) - if err == nil { - s.markSuccess(ctx, cand.ch.ID) - return resp, meta, nil - } - retry, penalize := switchable(err) - if !retry { - return nil, meta, err - } - if penalize { - s.markFailure(ctx, cand.ch.ID) - } - excluded[cand.ch.ID] = true - meta.Retries++ - lastErr = err - } - return nil, meta, lastErr + return routeRetry(ctx, s, req.Model, group, "EMBEDDING", func(cand *aiCandidate) (*aiwire.EmbeddingsResponse, error) { + return s.embedOnce(ctx, cand, req) + }) } // embedOnce 调用渠道向量化并装配 OpenAI 形态响应。 diff --git a/internal/service/webconsole.go b/internal/service/webconsole.go index a80a88a..a555675 100644 --- a/internal/service/webconsole.go +++ b/internal/service/webconsole.go @@ -46,6 +46,7 @@ type ConsoleSession struct { type ConsoleService struct { configs *OciConfigService pollInterval time.Duration // 清理残留连接的轮询间隔,测试注入缩短 + sessionTTL time.Duration // 会话回收检查周期,测试注入缩短 mu sync.Mutex sessions map[string]*ConsoleSession @@ -55,6 +56,7 @@ func NewConsoleService(configs *OciConfigService) *ConsoleService { return &ConsoleService{ configs: configs, pollInterval: 2 * time.Second, + sessionTTL: consoleSessionTTL, sessions: map[string]*ConsoleSession{}, } } @@ -161,11 +163,12 @@ func (s *ConsoleService) storeSession(cfgID uint, instanceID, region, typ, connI s.mu.Lock() s.sessions[sess.ID] = sess s.mu.Unlock() - time.AfterFunc(consoleSessionTTL, func() { s.expire(sess.ID) }) + time.AfterFunc(s.sessionTTL, func() { s.expire(sess.ID) }) return sess } -// expire TTL 到期回收:正在使用的会话跳过(连接断开后自然停止,无续期)。 +// expire TTL 到期回收:正在使用的会话跳过并重挂下一轮检查, +// 连接断开后由后续轮次回收,避免会话与云端连接常驻到进程退出。 func (s *ConsoleService) expire(id string) { s.mu.Lock() sess, ok := s.sessions[id] @@ -173,6 +176,7 @@ func (s *ConsoleService) expire(id string) { sess.mu.Lock() if sess.inUse { ok = false + time.AfterFunc(s.sessionTTL, func() { s.expire(id) }) } else { delete(s.sessions, id) } diff --git a/internal/service/webconsole_test.go b/internal/service/webconsole_test.go index a0e01fc..1252eac 100644 --- a/internal/service/webconsole_test.go +++ b/internal/service/webconsole_test.go @@ -116,3 +116,44 @@ func TestCreateConsoleSession(t *testing.T) { }) } } + +// TestConsoleSessionExpire 验证过期回收:在用会话跳过本轮,断开后下一轮回收。 +func TestConsoleSessionExpire(t *testing.T) { + tests := []struct { + name string + inUse bool + wantAlive bool + }{ + {name: "在用会话跳过回收", inUse: true, wantAlive: true}, + {name: "空闲会话回收并删云端连接", inUse: false, wantAlive: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := &consoleFakeClient{} + console, cfgID := newConsoleTestService(t, client) + sess, err := console.CreateSession(context.Background(), cfgID, "ocid1.instance.i", "", ConsoleTypeSerial) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + sess.MarkUse(tt.inUse) + console.expire(sess.ID) + if alive := console.Get(sess.ID) != nil; alive != tt.wantAlive { + t.Fatalf("session alive = %v, want %v", alive, tt.wantAlive) + } + if wantDel := !tt.wantAlive; (len(client.deleted) == 1) != wantDel { + t.Errorf("cloud connection deleted %v, want %v", client.deleted, wantDel) + } + if tt.inUse { + // 断开后下一轮回收(此前的缺陷:跳过后不再检查,会话常驻) + sess.MarkUse(false) + console.expire(sess.ID) + if console.Get(sess.ID) != nil { + t.Fatal("session still alive after idle expire round") + } + if len(client.deleted) != 1 { + t.Errorf("cloud connection not deleted after idle expire: %v", client.deleted) + } + } + }) + } +}