fix(gateway,history): REST 流式不再静默失败（G2）+ clientMsgId 幂等（G4）（协同 §G，均标「中高/中」）
- G2：`stream:true` 原经**有损总线**且**先回 200** ⇒ 总线饱和时客户端拿到 200 却什么都没发生。
改为**直达编排器**（与 ws chat 帧既有先例同一原则）：新文件 chat_stream_dispatch.go；200 变成有意义的
承诺（同步穿过「编排器接收 + 用户消息落库」才回）；顺带对齐 `202 {queued,queueDepth}` / `409 chat_busy`；
兜底路径（未接编排器才走总线）现在**返回错误**、ctx 与请求脱钩（WithoutCancel + 30min 上界）→ `503 chat_dispatch_failed`；
未接 loop 由 200 改为 `500 chat_unavailable`（与同步路径同口径，已同步改既有用例并新增钉住用例）
- G4：`chat`（WS 帧 + REST 体）新增**可选** `clientMsgId` ⇒ 幂等。
原子性两层，均非「先查再插」：① 同进程在 `o.mu` 同一临界区内完成「查认领表→查存储→预分配 id→认领」；
② 跨重启靠 `(session_id, client_msg_id)` **部分**唯一索引 + `INSERT … ON CONFLICT DO NOTHING`（命中则读回既有行）。
重复语义：REST `200 {duplicate:true, messageId}`；WS 新帧 `chat_duplicate`（owner/admin 档）；重复**不落库、不起回合**。
`clientMsgId` 回填进 DTO ⇒ `message_persisted` 与读接口都可见（客户端可据此对齐本地气泡，不再靠文本相等）
- 迁移 `0013_chat_client_msg_id`（up：加列 + 部分唯一索引 `idx_chat_messages_client_msg_id`；down：先删索引再删列），
已在 `data/cyrene.db` **副本**上验 up → 判重 → down → 再 up
- 拆分：store.go 647→498 行（`log_reads.go` 抽走四个读路径）；七个 SELECT 清单抽成 `messageColumns` 一处
- 复核：**全树 `go test ./... -count=1` 0 FAIL** + build/vet 绿 + linux/darwin 交叉编译绿；
四条真断言（总线饱和仍落库出帧且 `ws.chat` 零发布 / 同键两次仅一回合＋并发两条也仅一回合 / 省略键逐字不变 /
迁移副本 up-down-up）；**3 条变异反证**均已还原
- 越界说明（均已报备）：`history.go` DTO 字段（否则「读接口可见」不可能）、`policy_store_test.go`
（0013 使「回滚最后一步 == 0012」前提失效，改为按 schema_migrations 现算）、`extra_test.go`（补 runner）
- 顺带补交 P0-A 遗漏的新测试文件 `loop/memory_writeback_test.go`
This commit is contained in:
AskaEth committed
1 parent
da3a386daf
commit
7512d593c2
18 files changed
+1573
-216
No files matched your search
@@ -2,6 +2,7 @@ package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
@@ -55,10 +56,15 @@ func modeFrom(ctx context.Context) string {
|
||||
// 之后切换角色的会话仍保有准确的「谁说了这个」流。未接线
|
||||
// history 存储时它是空操作。被拒绝的回合（空、待处理已满）
|
||||
// 绝不触达它，因此不会持久化任何空内容。
|
||||
//
|
||||
// ctx 上若带着客户端的幂等键（G4，clientMsgHintFrom），第一条用户消息会带上它
|
||||
// 与编排器预分配的消息 id：写入因此可被 (session_id, client_msg_id) 的唯一索引
|
||||
// 判重。省略幂等键时 ctx 上没有该值，写入与加入这一列之前逐字相同。
|
||||
func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID, mode, personaID string) {
|
||||
if o.history == nil {
|
||||
return
|
||||
}
|
||||
hint := clientMsgHintFrom(ctx)
|
||||
for _, m := range messages {
|
||||
if m.Role != model.RoleUser {
|
||||
continue
|
||||
@@ -67,11 +73,28 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string
|
||||
if text == "" && image == "" {
|
||||
continue
|
||||
}
|
||||
full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID})
|
||||
msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID}
|
||||
// 幂等键只标注**第一条**用户消息：一个回合至多只有一条用户消息，而把
|
||||
// 同一个键盖到多条上会让它们互相冲突（唯一索引把它们判成同一条）。
|
||||
if hint.ClientMsgID != "" {
|
||||
msg.ClientMsgID, msg.ID = hint.ClientMsgID, hint.MessageID
|
||||
hint = clientMsgHint{}
|
||||
}
|
||||
full, err := o.history.AppendFull(ctx, msg)
|
||||
if err != nil {
|
||||
o.log.Error("orchestrator persist user", "err", err, "session", sessionID, "device_id", deviceID)
|
||||
// 落库失败就撤销认领：否则客户端拿同一个 clientMsgId 重试时会被判成
|
||||
// 「重复到达」而既不落库也不起回合 —— 这条消息从此再也进不来。
|
||||
if msg.ClientMsgID != "" {
|
||||
o.releaseClientMsg(sessionID, msg.ClientMsgID, msg.ID)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if msg.ClientMsgID != "" && full.ID != msg.ID {
|
||||
// 存储侧唯一索引裁决出既有行（例如本进程重启后另一次写入已落库）：
|
||||
// 把权威 id 覆盖进认领表，别让后续重复请求拿到一个悬空的 messageId。
|
||||
o.rememberClientMsg(sessionID, msg.ClientMsgID, full.ID)
|
||||
}
|
||||
o.notifyPersisted(full)
|
||||
}
|
||||
}
|
||||
@@ -111,3 +134,189 @@ func messageTextImage(m model.Message) (text, image string) {
|
||||
}
|
||||
return text, image
|
||||
}
|
||||
|
||||
// --- G4 客户端幂等键（clientMsgId）-------------------------------------------
|
||||
//
|
||||
// 全链路此前没有幂等：客户端超时重发同一段文本就是两条用户消息、两个回合。
|
||||
// 这一节给编排器加判重，与存储侧的 (session_id, client_msg_id) 部分唯一索引
|
||||
// 配合：编排器决定「这一轮还要不要跑」，存储决定「这一行还要不要写」。
|
||||
//
|
||||
// 两层职责刻意分开：内存认领表（sessionState.clientMsgs）让**同进程并发**只有一个
|
||||
// 赢家（这是唯一索引覆盖不到的窗口 —— 两次重发都还没插入），唯一索引则让判重
|
||||
// 跨重启依然成立（单二进制单进程，不存在跨进程并发）。
|
||||
|
||||
// maxClientMsgsPerSession 限制每会话保留在内存里的认领条数。淘汰最旧条目只会
|
||||
// 让一次很晚到达的重复多问一次存储，绝不会让重复被放行 —— 权威在存储。
|
||||
const maxClientMsgsPerSession = 64
|
||||
|
||||
// clientMsgHint 是一条幂等提示（G4）。ClientMsgID 是客户端自带的去重键
|
||||
// （空 = 老客户端，行为逐字不变）；MessageID 是编排器为这条消息**预分配**的
|
||||
// id，使并发的重复请求在第一条真正落库之前就能拿到一个确定（且终将存在）的
|
||||
// messageId 去对齐本地乐观气泡。
|
||||
type clientMsgHint struct {
|
||||
ClientMsgID string
|
||||
MessageID string
|
||||
}
|
||||
|
||||
// clientMsgHintKey 是携带 clientMsgHint 的 context 键。它从 HTTP/WS 边界
|
||||
// （REST 请求体 / ws chat 帧的 clientMsgId）引入编排器，与 deviceID、mode
|
||||
// 走同一种机制。
|
||||
type clientMsgHintKey struct{}
|
||||
|
||||
// withClientMsgHint 返回携带该提示的 context。空 ClientMsgID 是空操作：省略
|
||||
// clientMsgId 的请求在 ctx 上不留任何痕迹。
|
||||
func withClientMsgHint(ctx context.Context, h clientMsgHint) context.Context {
|
||||
if h.ClientMsgID == "" {
|
||||
return ctx
|
||||
}
|
||||
return context.WithValue(ctx, clientMsgHintKey{}, h)
|
||||
}
|
||||
|
||||
// clientMsgHintFrom 返回 ctx 携带的幂等提示，未设置时返回零值。
|
||||
func clientMsgHintFrom(ctx context.Context) clientMsgHint {
|
||||
h, _ := ctx.Value(clientMsgHintKey{}).(clientMsgHint)
|
||||
return h
|
||||
}
|
||||
|
||||
// clientMsgIDFrom 返回 ctx 携带的客户端幂等键，未设置时为 ""。
|
||||
func clientMsgIDFrom(ctx context.Context) string {
|
||||
return clientMsgHintFrom(ctx).ClientMsgID
|
||||
}
|
||||
|
||||
// clientMsgIDReader 是 HistoryStore 的一个**可选**更窄视图（与 persona_env.go
|
||||
// 的 sessionPersonaReader 同一做法）：按 (会话, clientMsgId) 定点读一条既有
|
||||
// 消息，供进程重启后的判重（内存认领表已随进程消失）。未实现该视图的存储替身
|
||||
// 或后端退回「内存认领表」判定，不因此报错，也无需为此实现新方法。
|
||||
type clientMsgIDReader interface {
|
||||
FindByClientMsgID(ctx context.Context, sessionID, clientMsgID string) (history.Message, error)
|
||||
}
|
||||
|
||||
// admit 是一次提交能否进入会话状态机的判定结果（design/15 §4 的队列容量 +
|
||||
// G4 的幂等判重）。两种为真的情形都意味着「不落库、不起回合」，但它们对客户端
|
||||
// 的含义相反：full 是「稍后重试」（队列满），dup 是「不必重试」（消息早已在库）。
|
||||
type admit struct {
|
||||
// full 报告该会话的待处理队列已满。
|
||||
full bool
|
||||
// dup 报告这次提交的 clientMsgId 已经到达过。
|
||||
dup bool
|
||||
// mode 是判定时的会话忙碌通道，供 PendingFull 事件如实报告（在锁内取值，
|
||||
// 解锁后再广播）。
|
||||
mode Mode
|
||||
// hint 是本次的幂等提示：dup 为真时指向既有/已认领的消息，否则携带本次
|
||||
// 落库要用的预分配 id（省略 clientMsgId 时是零值）。
|
||||
hint clientMsgHint
|
||||
}
|
||||
|
||||
// admitLocked 在状态锁下判定一次提交能否进入状态机，并在可以进入时完成 G4 的
|
||||
// 认领。它必须在 o.mu 下调用，且**不**在锁内广播任何状态事件 —— 调用方解锁后
|
||||
// 再发（与既有 PendingFull 路径同一纪律）。
|
||||
//
|
||||
// 顺序是刻意的：**队列满的拒绝排在认领之前**。被拒的回合没有落库，若在拒绝前
|
||||
// 认领了 clientMsgId，客户端用同一个键重试时会被判成「重复到达」，这条消息就
|
||||
// 再也进不来了 —— 那正是 G2 那类静默丢失的翻版。
|
||||
func (o *Orchestrator) admitLocked(ctx context.Context, st *sessionState, sessionID string) admit {
|
||||
if st.mode != ModeIdle && len(st.pending) >= o.maxPending {
|
||||
return admit{full: true, mode: st.mode}
|
||||
}
|
||||
hint, dup := o.claimClientMsgLocked(ctx, st, sessionID, clientMsgIDFrom(ctx))
|
||||
return admit{dup: dup, mode: st.mode, hint: hint}
|
||||
}
|
||||
|
||||
// claimClientMsgLocked 在状态锁下判重并认领一个 clientMsgId（G4）。duplicate 为
|
||||
// true 表示这条消息此前已经到达（hint.MessageID 指向既有/已认领的消息），调用方
|
||||
// 必须原地返回：既不落库、也不启动回合。duplicate 为 false 时 hint.MessageID 是
|
||||
// 本次落库要用的预分配 id（省略 clientMsgId 时 hint 与 duplicate 都是零值）。
|
||||
//
|
||||
// 必须在 o.mu 下调用：认领与状态读取必须是同一个临界区，否则两个并发的重复请求
|
||||
// 会双双认领成功、各起一个回合（这正是「先查再插」的经典竞态）。存储查询也是有界
|
||||
// 的定点读（部分唯一索引至多命中一行），与 startTurnMessages 的转录读取同一量级，
|
||||
// 因此不违反「状态锁不持有慢操作」。
|
||||
func (o *Orchestrator) claimClientMsgLocked(ctx context.Context, st *sessionState, sessionID, clientMsgID string) (clientMsgHint, bool) {
|
||||
if clientMsgID == "" {
|
||||
return clientMsgHint{}, false
|
||||
}
|
||||
if id, ok := st.clientMsgs[clientMsgID]; ok {
|
||||
return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, true
|
||||
}
|
||||
if id, ok := o.findPersistedClientMsg(ctx, sessionID, clientMsgID); ok {
|
||||
o.rememberClientMsgLocked(st, clientMsgID, id)
|
||||
return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, true
|
||||
}
|
||||
id, err := history.NewMessageID()
|
||||
if err != nil {
|
||||
// 生成失败就退回「不判重」：宁可多跑一轮重复，也不把这条消息挡在门外
|
||||
// —— 挡在门外就是静默丢失，正是本次要消灭的东西。
|
||||
o.log.Error("orchestrator client msg id", "err", err, "session", sessionID)
|
||||
return clientMsgHint{}, false
|
||||
}
|
||||
o.rememberClientMsgLocked(st, clientMsgID, id)
|
||||
return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, false
|
||||
}
|
||||
|
||||
// findPersistedClientMsg 报告该会话是否已有携带该 clientMsgId 的消息，是时返回
|
||||
// 其消息 id。存储未接、未实现该可选视图、或查询失败时一律报「没有」并记日志：
|
||||
// 判重失败的方向必须是**放行**（多一条重复），绝不是拒绝（丢一条消息）。
|
||||
func (o *Orchestrator) findPersistedClientMsg(ctx context.Context, sessionID, clientMsgID string) (string, bool) {
|
||||
reader, ok := o.history.(clientMsgIDReader)
|
||||
if !ok {
|
||||
return "", false
|
||||
}
|
||||
msg, err := reader.FindByClientMsgID(ctx, sessionID, clientMsgID)
|
||||
if err != nil {
|
||||
if !errors.Is(err, history.ErrMessageNotFound) {
|
||||
o.log.Error("orchestrator client msg lookup", "err", err, "session", sessionID)
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
return msg.ID, msg.ID != ""
|
||||
}
|
||||
|
||||
// rememberClientMsgLocked 认领一个 clientMsgId（必须在 o.mu 下调用），并按
|
||||
// maxClientMsgsPerSession 有界淘汰最旧的条目。
|
||||
func (o *Orchestrator) rememberClientMsgLocked(st *sessionState, clientMsgID, msgID string) {
|
||||
if st.clientMsgs == nil {
|
||||
st.clientMsgs = make(map[string]string, maxClientMsgsPerSession)
|
||||
}
|
||||
if _, ok := st.clientMsgs[clientMsgID]; !ok {
|
||||
st.clientMsgOrder = append(st.clientMsgOrder, clientMsgID)
|
||||
if len(st.clientMsgOrder) > maxClientMsgsPerSession {
|
||||
oldest := st.clientMsgOrder[0]
|
||||
st.clientMsgOrder = st.clientMsgOrder[1:]
|
||||
delete(st.clientMsgs, oldest)
|
||||
}
|
||||
}
|
||||
st.clientMsgs[clientMsgID] = msgID
|
||||
}
|
||||
|
||||
// rememberClientMsg 是 rememberClientMsgLocked 的加锁包装，供锁外的持久化路径
|
||||
// 在拿到存储裁决出的权威 id 后修正认领（见 persistUserMessages）。会话状态不
|
||||
// 存在时它是空操作（那意味着本次提交没有走过状态机，没有认领需要修正）。
|
||||
func (o *Orchestrator) rememberClientMsg(sessionID, clientMsgID, msgID string) {
|
||||
if clientMsgID == "" || msgID == "" {
|
||||
return
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if st := o.states[sessionID]; st != nil {
|
||||
o.rememberClientMsgLocked(st, clientMsgID, msgID)
|
||||
}
|
||||
}
|
||||
|
||||
// releaseClientMsg 撤销一次认领（落库失败时调用），使同一个 clientMsgId 的
|
||||
// 下一次重试仍然能落库，而不是被判重挡在门外、消息从此消失。它只撤销仍然属于
|
||||
// msgID 的那一条认领，不碰后来的认领者。
|
||||
func (o *Orchestrator) releaseClientMsg(sessionID, clientMsgID, msgID string) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
st := o.states[sessionID]
|
||||
if st == nil || st.clientMsgs[clientMsgID] != msgID {
|
||||
return
|
||||
}
|
||||
delete(st.clientMsgs, clientMsgID)
|
||||
for i, id := range st.clientMsgOrder {
|
||||
if id == clientMsgID {
|
||||
st.clientMsgOrder = append(st.clientMsgOrder[:i], st.clientMsgOrder[i+1:]...)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user