package events // 本文件是**保留期清扫**的验收(design/26 §4.4;切片 E1)。三类必测: // // 1. **≤0 = 关闭**(保留期 ≤0、间隔 ≤0 各一条):`Enabled()` 为 false 且 `Sweep` // 不动任何一行 —— 「配了却不生效」不存在(关闭原因由 NewStore 记 WARN); // 2. **正数清扫**:超期行按 `created_at < now − 保留期` 逐批删除,未超期的行与原样 // 保留,计数表**不参与 GC**; // 3. **失败不退出、不清空**:库不可用 / ctx 取消时 `Sweep` 返回错误(组合根记 WARN // 后等下一轮,绝不因此退出进程),**已经删的保持删除、未删的绝不被清空**。 import ( "context" "testing" "time" ) // seedExpired 写入一条指定时刻的 T2 事件(走真实写入路径,不直接拼 SQL)。 func seedExpired(t *testing.T, st *SQLStore, at time.Time, platformType string) { t.Helper() ev := sampleT2() ev.PlatformType = platformType ev.CreatedAt = at if err := st.Record(context.Background(), ev); err != nil { t.Fatalf("Record(%s): %v", platformType, err) } } // TestSweep_RetentionDisabled_边界:保留期 ≤0 ⇒ GC 关闭(显式),Sweep 是空操作。 func TestSweep_RetentionDisabled_边界(t *testing.T) { t.Parallel() ctx := context.Background() st, _ := newTestStore(t, func(cfg *Config) { cfg.GCInterval = time.Hour cfg.Retention = 0 }) if st.Enabled() { t.Fatal("Enabled() = true, want false(保留期 ≤0 = 关闭 GC)") } if st.Interval() != time.Hour || st.Retention() != 0 { t.Fatalf("Interval/Retention = %v/%v, want 1h/0", st.Interval(), st.Retention()) } seedExpired(t, st, testNow().Add(-1000*time.Hour), "group_recall") stats, err := st.Sweep(ctx) if err != nil { t.Fatalf("Sweep(disabled) err = %v, want nil", err) } if stats.Batches != 0 || stats.Deleted != 0 { t.Fatalf("Sweep(disabled) = %+v, want zero stats", stats) } if n, err := st.Count(ctx, Filter{}); err != nil { t.Fatalf("Count: %v", err) } else if n != 1 { t.Fatalf("rows = %d, want 1(GC 关闭时一行都不许动)", n) } } // TestSweep_IntervalDisabled_边界:间隔 ≤0 ⇒ 同样是关闭(与文档 GC 逐字同口径)。 func TestSweep_IntervalDisabled_边界(t *testing.T) { t.Parallel() ctx := context.Background() st, _ := newTestStore(t, func(cfg *Config) { cfg.GCInterval = 0 cfg.Retention = DefaultRetention }) if st.Enabled() { t.Fatal("Enabled() = true, want false(间隔 ≤0 = 关闭 GC)") } seedExpired(t, st, testNow().Add(-1000*time.Hour), "group_recall") stats, err := st.Sweep(ctx) if err != nil { t.Fatalf("Sweep(disabled) err = %v, want nil", err) } if stats.Deleted != 0 { t.Fatalf("Sweep(disabled) deleted = %d, want 0", stats.Deleted) } if n, err := st.Count(ctx, Filter{}); err != nil { t.Fatalf("Count: %v", err) } else if n != 1 { t.Fatalf("rows = %d, want 1", n) } } // TestSweep_DeletesExpiredOnly_正常:正数保留期 ⇒ 只删超期行;未超期的行保留, // **计数表不参与 GC**(design/26 §4.4:删了就答不上「上游推过没有」)。 func TestSweep_DeletesExpiredOnly_正常(t *testing.T) { t.Parallel() ctx := context.Background() const retention = 24 * time.Hour st, _ := newTestStore(t, func(cfg *Config) { cfg.GCInterval = time.Hour cfg.Retention = retention }) if !st.Enabled() { t.Fatal("Enabled() = false, want true") } seedExpired(t, st, testNow().Add(-48*time.Hour), "expired") seedExpired(t, st, testNow().Add(-time.Hour), "fresh") if err := st.Record(ctx, Event{ Platform: PlatformQQ, Tier: TierT1, Category: CategoryHeartbeat, PlatformType: "heartbeat", }); err != nil { t.Fatalf("Record(heartbeat): %v", err) } if err := st.Flush(ctx); err != nil { t.Fatalf("Flush: %v", err) } stats, err := st.Sweep(ctx) if err != nil { t.Fatalf("Sweep: %v", err) } if stats.Deleted != 1 { t.Fatalf("Deleted = %d, want 1", stats.Deleted) } rows, err := st.List(ctx, Filter{}) if err != nil { t.Fatalf("List: %v", err) } if len(rows) != 1 || rows[0].PlatformType != "fresh" { t.Fatalf("rows after sweep = %+v, want exactly the fresh one", rows) } // 计数表不参与 GC:即便某个键的**事件行**被清扫掉,它的计数行照旧在(design/26 // §4.4 —— 删了就再也答不上「上游推过没有」)。 counters := mustCounters(t, st, PlatformQQ) seen := make(map[string]int64, len(counters)) for _, c := range counters { seen[c.PlatformType] = c.Count } for _, key := range []string{"expired", "fresh", "heartbeat"} { if seen[key] != 1 { t.Fatalf("counter %q = %d, want 1(全部计数行都必须留着): %+v", key, seen[key], counters) } } } // TestSweep_MultipleBatches_边界:积压超过单批上限时按批删除(一轮的时长与内存占用 // 有界),且一轮内能把超过一批的量扫完。 func TestSweep_MultipleBatches_边界(t *testing.T) { t.Parallel() ctx := context.Background() st, _ := newTestStore(t, func(cfg *Config) { cfg.GCInterval = time.Hour cfg.Retention = time.Hour }) const rows = sweepBatchSize + 1 old := testNow().Add(-100 * time.Hour) for i := 0; i < rows; i++ { seedExpired(t, st, old, "expired") } stats, err := st.Sweep(ctx) if err != nil { t.Fatalf("Sweep: %v", err) } if stats.Deleted != int64(rows) { t.Fatalf("Deleted = %d, want %d", stats.Deleted, rows) } if stats.Batches != 2 { t.Fatalf("Batches = %d, want 2 (单批上限 %d)", stats.Batches, sweepBatchSize) } if n, err := st.Count(ctx, Filter{}); err != nil { t.Fatalf("Count: %v", err) } else if n != 0 { t.Fatalf("rows = %d, want 0", n) } } // TestSweep_FailureDoesNotClear_异常:库不可用时 Sweep 返回错误(组合根记 WARN 后 // 继续),ctx 取消时立刻停下且**不清空**未删的行。 func TestSweep_FailureDoesNotClear_异常(t *testing.T) { t.Parallel() ctx := context.Background() st, conn := newTestStore(t, func(cfg *Config) { cfg.GCInterval = time.Hour cfg.Retention = time.Hour }) seedExpired(t, st, testNow().Add(-100*time.Hour), "expired") // ① ctx 取消:立即停下、返回错误、一行都不删(未删的绝不因此被清空)。 canceled, cancel := context.WithCancel(ctx) cancel() stats, err := st.Sweep(canceled) if err == nil { t.Fatal("Sweep(canceled ctx) err = nil, want an error") } if stats.Deleted != 0 { t.Fatalf("Sweep(canceled ctx) deleted = %d, want 0", stats.Deleted) } if n, cerr := st.Count(ctx, Filter{}); cerr != nil { t.Fatalf("Count: %v", cerr) } else if n != 1 { t.Fatalf("rows = %d, want 1 (取消不得清空)", n) } // ② 库不可用:同样返回错误(调用方等下一轮),且不 panic。 if err := conn.Close(); err != nil { t.Fatalf("关闭内存库: %v", err) } if _, err := st.Sweep(ctx); err == nil { t.Fatal("Sweep(closed db) err = nil, want an error") } }