AskaEth
7512d593c2
- 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`
112 lines
4.2 KiB
Go
112 lines
4.2 KiB
Go
package history
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"testing"
|
||
)
|
||
|
||
// 本文件锁定 Android 协同 §G4 的存储半边：客户端幂等键（client_msg_id）由
|
||
// (session_id, client_msg_id) 的**部分唯一索引**原子裁决，因此判重不是「先查
|
||
// 再插」的竞态写法，而是「写入被数据库拒绝，然后读回既有那一行」。
|
||
// 部分（WHERE client_msg_id != ''）是必须的：否则同一会话的所有空键行会互相
|
||
// 冲突，等于禁止一个会话出现第二条消息。
|
||
|
||
func TestAppendFullDedupesClientMsgID(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
first, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "同一段", ClientMsgID: "cm-1"})
|
||
if err != nil {
|
||
t.Fatalf("第一次写入：%v", err)
|
||
}
|
||
// 第二次：换一个 id（像客户端重发时服务端会另铸一个），同一个 clientMsgId。
|
||
second, err := st.AppendFull(ctx, Message{ID: "重发时的新 id", SessionID: MainSessionID, Role: RoleUser, Content: "同一段", ClientMsgID: "cm-1"})
|
||
if err != nil {
|
||
t.Fatalf("第二次写入：%v", err)
|
||
}
|
||
if second.ID != first.ID {
|
||
t.Fatalf("第二次返回的 id = %q，want 既有那条 %q（唯一索引裁决，不新写一行）", second.ID, first.ID)
|
||
}
|
||
if second.ClientMsgID != "cm-1" {
|
||
t.Fatalf("第二次返回的消息 = %+v，want 回填的 clientMsgId", second)
|
||
}
|
||
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 50, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 1 {
|
||
t.Fatalf("行数 = %d, want 1（同一个 clientMsgId 只落一条）", len(msgs))
|
||
}
|
||
if msgs[0].ClientMsgID != "cm-1" {
|
||
t.Fatalf("读回的消息 = %+v，want clientMsgId=cm-1（读接口可见）", msgs[0])
|
||
}
|
||
|
||
// 省略幂等键（空串）不参与约束：同一会话的两条空键消息都必须写得进去。
|
||
for i := 0; i < 2; i++ {
|
||
if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "无键"}); err != nil {
|
||
t.Fatalf("第 %d 条空键写入：%v", i+1, err)
|
||
}
|
||
}
|
||
msgs, err = st.List(ctx, MainSessionID, 0, 50, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 3 {
|
||
t.Fatalf("行数 = %d, want 3（空 clientMsgId 绝不受唯一索引约束）", len(msgs))
|
||
}
|
||
}
|
||
|
||
func TestAppendDedupesClientMsgIDPerSession(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
// 幂等键只在**会话内**唯一：两个会话用同一个键互不影响
|
||
// （client_msg_id 由客户端生成，不同会话之间没有可比性）。
|
||
main, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "主会话", ClientMsgID: "cm-x"})
|
||
if err != nil {
|
||
t.Fatalf("主会话写入：%v", err)
|
||
}
|
||
sub := MainSessionID + ":sub-1"
|
||
other, err := st.AppendFull(ctx, Message{SessionID: sub, Role: RoleUser, Content: "子会话", ClientMsgID: "cm-x"})
|
||
if err != nil {
|
||
t.Fatalf("子会话写入：%v", err)
|
||
}
|
||
if other.ID == main.ID {
|
||
t.Fatalf("子会话返回了主会话的行 %q，want 各自一条", main.ID)
|
||
}
|
||
}
|
||
|
||
func TestFindByClientMsgID(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "定位我", ClientMsgID: "cm-find"})
|
||
if err != nil {
|
||
t.Fatalf("写入：%v", err)
|
||
}
|
||
got, err := st.FindByClientMsgID(ctx, MainSessionID, "cm-find")
|
||
if err != nil {
|
||
t.Fatalf("按幂等键读取：%v", err)
|
||
}
|
||
if got.ID != full.ID || got.ClientMsgID != "cm-find" {
|
||
t.Fatalf("读回 = %+v，want id=%q clientMsgId=cm-find", got, full.ID)
|
||
}
|
||
|
||
// 没见过的键与空键都必须是明确的 ErrMessageNotFound（编排器据此判定「不是
|
||
// 重复」，而不是把查询错误当成重复）。
|
||
for _, key := range []string{"cm-missing", ""} {
|
||
if _, err := st.FindByClientMsgID(ctx, MainSessionID, key); !errors.Is(err, ErrMessageNotFound) {
|
||
t.Fatalf("键 %q 的错误 = %v，want wrap of ErrMessageNotFound", key, err)
|
||
}
|
||
}
|
||
// 会话也必须匹配：同一个键在别的会话里不存在。
|
||
if _, err := st.FindByClientMsgID(ctx, MainSessionID+":other", "cm-find"); !errors.Is(err, ErrMessageNotFound) {
|
||
t.Fatalf("跨会话查询的错误 = %v，want wrap of ErrMessageNotFound", err)
|
||
}
|
||
}
|