换手机拉三年记录，数据库被打满
「回声」支持消息漫游。用户换手机后，客户端会拉取历史消息。最初接口没有分页，默认返回该会话全部记录。普通用户只有几千条，测试没问题；一位用了三年的重度用户，单个会话有 20 多万条消息。
案例假设值：这次请求返回约 200 MB 数据，序列化和网络传输持续了十几秒，数据库连接被占用，连接池很快耗尽，同一实例上的发送接口开始超时。用户端看到的是“一直转圈”，重试又发起同样的全量查询，形成雪崩。
问题不是“历史消息太多”，而是把在线读路径当成了离线导出：没有游标、没有条数上限、没有冷热分离。消息存储和漫游必须一起设计。
按会话分片、按时间分区
消息的访问模式非常集中：绝大多数请求只读某个会话的最新一页，少数请求按时间往回翻。存储布局应该围绕这个模式设计。
| 布局 | 分片键或分区键 | 优点 | 代价 | 适用 |
|---|---|---|---|---|
| 按会话分片 | conv_id | 单会话读写集中，顺序好 | 大群或热会话倾斜 | 主路径 |
| 按时间分区 | created_at | 冷热分离，过期删除便宜 | 跨时间查询要合并 | 归档与清理 |
| 按用户分片 | user_id | 收件箱查询快 | 群消息写扩散放大 | 收件箱表 |
| 全局单表 | 无 | 实现简单 | 数据量一大就不可维护 | 原型 |
会话分片解决什么
按 conv_id 分片，能让同一会话的消息落在同一节点，序号分配、分页和已读游标都可以局部完成。缺点是热会话：一个万人群的消息量可能超过几万个普通单聊，分片键相同就会形成热点。常见缓解是把超大群拆成子分区，或者把群消息的存储与成员收件箱分开。
时间分区解决什么
按时间分区让“删除三年前的消息”变成删除整个分区，而不是逐行 DELETE。它也让冷数据可以放到更便宜的存储。代价是漫游跨时间范围时要合并多个分区，所以查询必须带时间或序号边界，不能无条件扫描。
在线路径只读最新一页
在线漫游接口应该只做一件事：给定会话和游标，返回固定条数。默认不带游标时返回最新一页；用户往上翻时，用上一页最小 Seq 作为 before。任何“返回全部历史”的参数都不应该存在。
失败场景：没有游标的全量漫游
那次事故的接口签名大致是 GET /conversations/<built-in function id>/messages，没有 limit，也没有 before。客户端换机时调用它，服务端把该会话所有消息查出来，按时间升序返回。
后果是连锁的：
- 单次查询扫描 20 多万行，占用数据库连接十几秒。
- 结果集在应用层组装成切片，内存峰值超过实例限制，触发 GC 抖动。
- 客户端解析 200 MB JSON，低端机直接闪退。
- 用户重试放大查询，连接池耗尽，发送和已读接口一起超时。
修复不是简单加一个 LIMIT 100。如果排序和游标不对，用户往上翻时会重复或跳过消息。正确做法是固定排序为 Seq DESC，用 before 作为游标，返回后再由客户端反转展示；同时把保留策略落到存储层，过期数据不再参与查询。
Go 实现：保留裁剪与漫游分页
Keep 同时处理时间保留和每会话条数上限，RoamPage 处理游标分页。两个函数都不修改入参，方便在写入、归档和测试之间复用。
// file: internal/im/retention.go
// tier: L1
// verify: go test ./internal/im/
package im
import (
"fmt"
"sort"
)
const secondsPerDay = 86400
// Policy 是单个会话的消息保留策略。
type Policy struct {
RetentionDays int
MaxPerConv int
}
// Keep 同时按时间和条数裁剪消息。它按会话分组，保留最新 MaxPerConv 条，
// 并丢弃早于 now-RetentionDays 的消息。返回值是新切片，入参不被修改。
// 结果按会话 ID、再按 Seq 升序排列，便于下游做稳定 diff。
func Keep(messages []Message, now int64, p Policy) ([]Message, error) {
if p.RetentionDays <= 0 {
return nil, fmt.Errorf("im: retention days %d must be positive", p.RetentionDays)
}
if p.MaxPerConv <= 0 {
return nil, fmt.Errorf("im: max per conversation %d must be positive", p.MaxPerConv)
}
if now <= 0 {
return nil, fmt.Errorf("im: now %d must be positive", now)
}
cutoff := now - int64(p.RetentionDays)*secondsPerDay
byConv := make(map[string][]Message)
for i, m := range messages {
if m.ConvID == "" {
return nil, fmt.Errorf("im: message %d has empty conversation id", i)
}
if m.Seq <= 0 {
return nil, fmt.Errorf("im: message %q has non-positive seq", m.ID)
}
if m.At < cutoff {
continue
}
byConv[m.ConvID] = append(byConv[m.ConvID], m)
}
kept := make([]Message, 0, len(messages))
for _, conv := range sortedKeys(byConv) {
msgs := byConv[conv]
sort.SliceStable(msgs, func(i, j int) bool {
if msgs[i].Seq != msgs[j].Seq {
return msgs[i].Seq < msgs[j].Seq
}
if msgs[i].At != msgs[j].At {
return msgs[i].At < msgs[j].At
}
return msgs[i].ID < msgs[j].ID
})
if len(msgs) > p.MaxPerConv {
msgs = msgs[len(msgs)-p.MaxPerConv:]
}
kept = append(kept, msgs...)
}
return kept, nil
}
// RoamPage 返回 Seq 小于 before 的最新 limit 条消息，按 Seq 倒序排列。
// before <= 0 表示从最新一条开始。调用方应只传入同一会话的消息。
func RoamPage(messages []Message, before int64, limit int) ([]Message, error) {
if before < 0 {
return nil, fmt.Errorf("im: before %d must not be negative", before)
}
if limit <= 0 {
return nil, fmt.Errorf("im: limit %d must be positive", limit)
}
copied := make([]Message, 0, len(messages))
for i, m := range messages {
if m.ID == "" || m.ConvID == "" {
return nil, fmt.Errorf("im: message %d has empty identity field", i)
}
if m.Seq <= 0 {
return nil, fmt.Errorf("im: message %q has non-positive seq", m.ID)
}
if before > 0 && m.Seq >= before {
continue
}
copied = append(copied, m)
}
sort.SliceStable(copied, func(i, j int) bool {
if copied[i].Seq != copied[j].Seq {
return copied[i].Seq > copied[j].Seq
}
return copied[i].ID > copied[j].ID
})
if len(copied) > limit {
copied = copied[:limit]
}
return copied, nil
}
func sortedKeys(m map[string][]Message) []string {
keys := make([]string, 0, len(m))
for k := range m {
keys = append(keys, k)
}
sort.Strings(keys)
return keys
}
几个设计取舍：
- 保留期按天配置，内部换算成秒。
At使用 Unix 秒，边界比较简单；如果业务需要毫秒，应该在入口统一换算，而不是让两个函数各自猜单位。 Keep先按ConvID分组，再在组内按Seq排序，保留最后MaxPerConv条。这样“最新 N 条”由序号决定，不受输入顺序影响。- 时间过期和条数上限是“或”的关系：任一条件满足就裁掉。只做时间保留，热会话仍然可能积累过多消息；只做条数上限，冷会话的旧数据永远删不掉。
RoamPage在before > 0时过滤Seq >= before，保证游标严格向前。before <= 0表示从最新一条开始，方便首次进入会话。- 返回结果按
Seq倒序，客户端可以直接渲染“最新在上”；如果需要正序，应在展示层反转，而不是让服务端返回两种顺序。
测试：过期、超限与游标边界
测试覆盖三个容易出错的边界：保留期边界、每会话上限、游标是否严格向前。
// file: internal/im/retention_test.go
// tier: L1
// verify: go test ./internal/im/
package im
import "testing"
func TestKeepDropsExpiredMessages(t *testing.T) {
now := int64(10 * secondsPerDay)
msgs := []Message{
{ID: "m1", ConvID: "c1", Seq: 1, At: 6 * secondsPerDay},
{ID: "m2", ConvID: "c1", Seq: 2, At: 7 * secondsPerDay},
{ID: "m3", ConvID: "c2", Seq: 1, At: 9 * secondsPerDay},
}
got, err := Keep(msgs, now, Policy{RetentionDays: 3, MaxPerConv: 100})
if err != nil {
t.Fatalf("Keep() error = %v", err)
}
if len(got) != 2 || got[0].ID != "m2" || got[1].ID != "m3" {
t.Fatalf("Keep() = %+v, want [m2 m3]", got)
}
}
func TestKeepCapsPerConversation(t *testing.T) {
now := int64(10 * secondsPerDay)
var msgs []Message
for i := 1; i <= 4; i++ {
msgs = append(msgs, Message{ID: "a" + string(rune('0'+i)), ConvID: "c1", Seq: int64(i), At: now})
}
msgs = append(msgs, Message{ID: "b1", ConvID: "c2", Seq: 1, At: now})
got, err := Keep(msgs, now, Policy{RetentionDays: 30, MaxPerConv: 2})
if err != nil {
t.Fatalf("Keep() error = %v", err)
}
if len(got) != 3 {
t.Fatalf("Keep() kept %d messages, want 3", len(got))
}
if got[0].ID != "a3" || got[1].ID != "a4" || got[2].ID != "b1" {
t.Fatalf("Keep() = %+v, want [a3 a4 b1]", got)
}
}
func TestKeepDoesNotMutateInput(t *testing.T) {
now := int64(10 * secondsPerDay)
msgs := []Message{
{ID: "m2", ConvID: "c1", Seq: 2, At: now},
{ID: "m1", ConvID: "c1", Seq: 1, At: now},
}
got, err := Keep(msgs, now, Policy{RetentionDays: 1, MaxPerConv: 1})
if err != nil {
t.Fatalf("Keep() error = %v", err)
}
if len(got) != 1 || got[0].ID != "m2" {
t.Fatalf("Keep() = %+v, want [m2]", got)
}
if msgs[0].ID != "m2" || msgs[1].ID != "m1" {
t.Fatalf("Keep() mutated input order: %+v", msgs)
}
}
func TestKeepRejectsBadPolicy(t *testing.T) {
msgs := []Message{{ID: "m1", ConvID: "c1", Seq: 1, At: 100}}
cases := []Policy{
{RetentionDays: 0, MaxPerConv: 10},
{RetentionDays: 7, MaxPerConv: 0},
}
for _, p := range cases {
if _, err := Keep(msgs, 1000, p); err == nil {
t.Fatalf("Keep(%+v) = nil error", p)
}
}
if _, err := Keep(msgs, 0, Policy{RetentionDays: 7, MaxPerConv: 10}); err == nil {
t.Fatal("Keep(now=0) = nil error")
}
}
func TestRoamPageUsesDescendingCursor(t *testing.T) {
msgs := []Message{
{ID: "m3", ConvID: "c1", Seq: 3, At: 3},
{ID: "m1", ConvID: "c1", Seq: 1, At: 1},
{ID: "m5", ConvID: "c1", Seq: 5, At: 5},
{ID: "m4", ConvID: "c1", Seq: 4, At: 4},
{ID: "m2", ConvID: "c1", Seq: 2, At: 2},
}
cases := []struct {
before int64
limit int
want []string
}{
{5, 2, []string{"m4", "m3"}},
{3, 2, []string{"m2", "m1"}},
{0, 2, []string{"m5", "m4"}},
{1, 2, []string{}},
}
for _, tc := range cases {
got, err := RoamPage(msgs, tc.before, tc.limit)
if err != nil {
t.Fatalf("RoamPage(before=%d) error = %v", tc.before, err)
}
if len(got) != len(tc.want) {
t.Fatalf("RoamPage(before=%d) = %+v, want %v", tc.before, got, tc.want)
}
for i := range tc.want {
if got[i].ID != tc.want[i] {
t.Fatalf("RoamPage(before=%d)[%d] = %q, want %q", tc.before, i, got[i].ID, tc.want[i])
}
}
}
}
func TestRoamPageRejectsBadArgs(t *testing.T) {
msgs := []Message{{ID: "m1", ConvID: "c1", Seq: 1, At: 1}}
if _, err := RoamPage(msgs, -1, 10); err == nil {
t.Fatal("RoamPage(before=-1) = nil error")
}
if _, err := RoamPage(msgs, 0, 0); err == nil {
t.Fatal("RoamPage(limit=0) = nil error")
}
}
TestKeepDropsExpiredMessages 把一条消息放在保留期边界上，确认 At == cutoff 仍然保留；更早的消息被裁掉。边界包含还是排除，必须在代码和文档里保持一致。
TestKeepCapsPerConversation 验证每个会话独立保留最新 2 条，不会因为另一个会话消息少就多保留。结果按会话 ID 排序，便于稳定比较。
TestKeepDoesNotMutateInput 确认裁剪不会原地重排入参。写入路径可能需要用原始切片做审计，函数擅自排序会破坏调用方假设。
TestRoamPageUsesDescendingCursor 覆盖从最新开始、带游标、游标到头部三种情况，并检查 Seq >= before 被排除。TestRoamPageRejectsBadArgs 覆盖负游标和 0 条数。
运行命令与真实输出
cd .local/checks/webapp-lab && go test ./internal/im/ -run '^(TestKeepDropsExpiredMessages|TestKeepCapsPerConversation|TestKeepDoesNotMutateInput|TestKeepRejectsBadPolicy|TestRoamPageUsesDescendingCursor|TestRoamPageRejectsBadArgs)$' -count=1 -v
一次本机运行的真实输出如下，耗时随机器变化：
=== RUN TestKeepDropsExpiredMessages
--- PASS: TestKeepDropsExpiredMessages (0.00s)
=== RUN TestKeepCapsPerConversation
--- PASS: TestKeepCapsPerConversation (0.00s)
=== RUN TestKeepDoesNotMutateInput
--- PASS: TestKeepDoesNotMutateInput (0.00s)
=== RUN TestKeepRejectsBadPolicy
--- PASS: TestKeepRejectsBadPolicy (0.00s)
=== RUN TestRoamPageUsesDescendingCursor
--- PASS: TestRoamPageUsesDescendingCursor (0.00s)
=== RUN TestRoamPageRejectsBadArgs
--- PASS: TestRoamPageRejectsBadArgs (0.00s)
PASS
ok webapp/internal/im 0.277s
PASS 表示裁剪和分页语义都符合预期。测试是纯内存的，真实系统的耗时主要取决于分区裁剪、索引和网络；但只要游标和排序规则一致，分页结果就可以逐页稳定复现。
小结
消息存储要同时回答两个问题：数据放在哪里，在线请求读多少。按会话分片让单会话读写局部化，按时间分区让冷热分离和过期删除变便宜；漫游接口必须有游标和条数上限，不能返回全量历史。
Policy、Keep 和 RoamPage 把保留与分页规则固定下来：过期或超限的消息被裁掉，每会话保留最新 N 条，漫游按 Seq 倒序且游标严格向前。下一章进入实时投递，讨论在线、离线与多端同步。
阅读导航：上一节：8.2 单聊、群聊与系统消息 · 下一节：9.1 在线投递与离线补偿 。
常见错误与边界
- 漫游接口没有
limit。只要有一个重度用户，单次查询就可能拖垮连接池。 - 用 offset 分页。消息持续写入时 offset 会漂移，翻页出现重复或漏读；应该用 Seq 游标。
- 游标使用
Seq <= before。这样上一页的最后一条会在下一页重复出现，客户端需要额外去重。 - 时间单位和存储不一致。
At是秒、配置是毫秒，保留期会差 1000 倍。 - 只做时间保留，不做条数上限。热会话在保留期内仍可能积累几十万条消息。
- 只做条数上限，不做时间保留。冷会话的旧数据永远不会被清理，存储成本只增不减。
- 在线查询扫描全部分区。时间分区只有在查询带边界时才有裁剪效果。
- 在请求链路里同步清理过期数据。删除应该由后台任务或分区操作完成，不能混进读路径。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。