JianFeeeee
9d4718a412
## SSE Last-Event-ID 补投 EventSource 断线重连时自带 Last-Event-ID 头，但服务端直接忽略了—— 所有断线期间的邮件通知都丢失。用户刷新页面也会错过已推的事件。 改为 per-user 事件环形缓冲区（500 条，~100KB/用户，20 在线 ≈ 2MB）： 每次 Broadcast/SendToUser/SendToAgent 同时写入对应用户的缓冲区； AddClient 时取 Last-Event-ID 头，找到该 ID 的位置后从下一条回放。 找不到 ID 说明事件已被覆盖（缓冲区溢出），从头回放全部。 事件 ID 用全局递增序列号（非 UUID），EventSource 的 Last-Event-ID 就是靠这个 ID 记住断点的。 新增测试：缓冲区回放、溢出行为、并发安全（10 goroutine × 200 次 push）、 端到端重连验证（SendToUser → 带 Last-Event-ID 的 AddClient → 补投）。 ## 连接状态指示器 Sidebar 用户头像右下角的小圆点：绿=已连接，黄=连接中，橙=重连中，红=断开。 NarrowNav 底栏也有（移动端）。 SSE 模块新增 onSSEStatus/getSSEStatus 接口，onerror/onopen 驱动状态变化。 状态点用 absolute 定位在头像边缘，不遮挡文字。 ## 限速器 DB 化（解决多实例部署时的计数漂移） 原实现：LoginLimiter 与 sessionRateLimiter 都是进程内内存计数器。 多实例部署时各自独立计数，等效上限变成 N 倍。 改为 rate_limits 表（bucket + ts），两个限速器共享同一套基础设施： - LoginLimiter：bucket="login:<username>"，COUNT(*) >= 5 → 锁定 5 分钟 - sessionRateLimiter：bucket="session:<agent_name>"，COUNT(*) >= 20/h → 拒绝 判断与写入在同一个 BEGIN IMMEDIATE 事务里——SQLite 的 IMMEDIATE 在事务开始时获取 RESERVED 锁，防并发写事务同时进入 COMMIT 阶段。 实测 80 并发下恰好放行 20 次（旧内存版同样通过，但 DB 版才能多实例共享）。 DB 不可用时放行（宁可放开限速也不能让用户完全无法使用）。 新建 rate_limits 表迁移（SQLite + PG 两版）。
101 lines
2.4 KiB
Go
101 lines
2.4 KiB
Go
package sse
|
||
|
||
import (
|
||
"net/http/httptest"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestEventRingPushReplay(t *testing.T) {
|
||
ring := newEventRing(5)
|
||
|
||
// 推 3 条
|
||
for i := 1; i <= 3; i++ {
|
||
ring.push(StoredEvent{
|
||
ID: string(rune('0' + i)),
|
||
EventType: "test",
|
||
Data: []byte(`{"n":` + string(rune('0'+i)) + `}`),
|
||
Timestamp: time.Now(),
|
||
})
|
||
}
|
||
|
||
// 空 afterID → 首次连接，不回放（缓冲区未满）
|
||
rec := httptest.NewRecorder()
|
||
ring.replay("", rec, rec)
|
||
if rec.Body.Len() > 0 {
|
||
t.Error("首次连接不应回放事件，实际:", rec.Body.String())
|
||
}
|
||
|
||
// 有 afterID → 从下一条开始回放
|
||
rec2 := httptest.NewRecorder()
|
||
ring.replay("1", rec2, rec2)
|
||
body := rec2.Body.String()
|
||
if !strings.Contains(body, "id: 2") {
|
||
t.Error("afterID=1 应该回放 id:2，实际:", body)
|
||
}
|
||
if !strings.Contains(body, "id: 3") {
|
||
t.Error("afterID=1 应该回放 id:3，实际:", body)
|
||
}
|
||
if strings.Contains(body, "id: 1") {
|
||
t.Error("afterID=1 不应回放 id:1，实际:", body)
|
||
}
|
||
|
||
// 不存在的 afterID → 从头回放全部
|
||
rec3 := httptest.NewRecorder()
|
||
ring.replay("999", rec3, rec3)
|
||
body3 := rec3.Body.String()
|
||
if !strings.Contains(body3, "id: 1") {
|
||
t.Error("不存在的 afterID 应从头回放，实际:", body3)
|
||
}
|
||
}
|
||
|
||
func TestEventRingOverflow(t *testing.T) {
|
||
ring := newEventRing(3)
|
||
|
||
// 推 5 条（超过容量 3，最旧的 2 条被覆盖）
|
||
for i := 1; i <= 5; i++ {
|
||
ring.push(StoredEvent{
|
||
ID: string(rune('0' + i)),
|
||
EventType: "test",
|
||
Data: []byte(`{}`),
|
||
Timestamp: time.Now(),
|
||
})
|
||
}
|
||
|
||
if !ring.full {
|
||
t.Fatal("推了 5 条进容量 3 的缓冲区，应该已满")
|
||
}
|
||
|
||
// afterID=2 已被覆盖 → 找不到位置，从头回放全部
|
||
rec := httptest.NewRecorder()
|
||
ring.replay("2", rec, rec)
|
||
body := rec.Body.String()
|
||
if !strings.Contains(body, "id: 3") || !strings.Contains(body, "id: 5") {
|
||
t.Error("缓冲区溢出后应能回放可用范围，实际:", body)
|
||
}
|
||
}
|
||
|
||
func TestEventRingConcurrent(t *testing.T) {
|
||
ring := newEventRing(100)
|
||
|
||
done := make(chan bool, 10)
|
||
for i := 0; i < 10; i++ {
|
||
go func() {
|
||
for j := 0; j < 200; j++ {
|
||
ring.push(StoredEvent{
|
||
ID: "evt",
|
||
EventType: "test",
|
||
Data: []byte(`{}`),
|
||
Timestamp: time.Now(),
|
||
})
|
||
}
|
||
done <- true
|
||
}()
|
||
}
|
||
for i := 0; i < 10; i++ {
|
||
<-done
|
||
}
|
||
// 只验证不 panic，不验证内容（并发下顺序无意义）
|
||
}
|