From 44c25eb03bcd690f8e161b9001bf180b755dc315 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 13 Oct 2026 21:35:47 +0800 Subject: [PATCH] =?UTF-8?q?refactor(memory):=20=E5=88=B0=E6=9C=9F=E6=B8=85?= =?UTF-8?q?=E6=89=AB=E6=8C=82=E7=82=B9=E4=BB=8E=20wait=20=E6=B3=A8?= =?UTF-8?q?=E5=85=A5=E7=BC=9D=E6=94=B9=E4=B8=BA=20TidyService.Start?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 塔裁定：清扫不许挂在注入缝上 —— 包一层 s.wait 会让「清扫是否存活」依赖于 「loop 每一轮都调 s.wait」这个**没有任何契约固定**的假设；将来谁换了等待原语、 或把循环改成 select + 自己的 timer，清扫会**静默消失**（不报错、不红测试， 只是临时期限再也删不掉）。后台件起在 Start 里，读者一眼看到它们同生共死。 - tidy_service.go：Start 里紧跟 go s.loop(cctx) 起 go s.expirySweepLoop(cctx)， wg 计数与既有那一笔同处（同一把锁内加，避免 Add 与 Stop 的 wg.Wait 相撞）。 `if !s.cfg.Enabled { … return }` 仍在其前 ⇒ 语义与改动前逐字相同：整理被关掉 ⇒ 清扫不跑（与 preinsert_gc 随审查服务启停同口径）。 - expire_loop.go：删掉 AttachExpirySweep 与对 s.wait 的包裹；文件头写明 「挂点必须是 Start 里那处显式的 go」，防止后人改回注入缝。 - expire_loop_test.go：那条运行时对照改写成 TestTidyServiceStart_清扫随服务起停 —— 真 Start、真 Stop（Stop 不超时即证明清扫随 ctx 退出且登记在同一个 wg）。 - wire_memory_tidy.go：去掉组合根的挂载调用，注释改成「起在 Start 里，组合根零接线」。 - wire_memory_tidy_test.go：把 go s.expirySweepLoop(cctx) 钉进 TestMemoryTidyIsWired 的 seams（摘掉那一行 ⇒ 用例红，已实测）。 --- cmd/cyrene/wire_memory_tidy.go | 15 +++--- cmd/cyrene/wire_memory_tidy_test.go | 12 +++++ internal/runtime/memory/expire_loop.go | 55 ++++---------------- internal/runtime/memory/expire_loop_test.go | 57 +++++++++++---------- internal/runtime/memory/tidy_service.go | 15 ++++++ 5 files changed, 73 insertions(+), 81 deletions(-) diff --git a/cmd/cyrene/wire_memory_tidy.go b/cmd/cyrene/wire_memory_tidy.go index de02536..6cb278b 100644 --- a/cmd/cyrene/wire_memory_tidy.go +++ b/cmd/cyrene/wire_memory_tidy.go @@ -29,9 +29,9 @@ import ( // 使「面板优先」从第一次启动起就成立，而不是靠 `.env` 一直兜着； // - **到期清扫的挂点**（2026-10-13 追加，切片 M3，design/23 §19）：临时记忆到期 // 后要**真删掉**（读侧时效闸只负责「看不见」）。清扫是 memory 包内的后台件 -// （expire.go / expire_loop.go），这里只把它挂到整理服务的生命周期上 -// （AttachExpirySweep）—— 它随心跳 ctx 跑、随 Stop 停，因此不新开关闭链、 -// 不新开端口、不加 env 键。 +// （expire.go / expire_loop.go），起在 **TidyService.Start** 里、与心跳共用 +// 同一个 ctx 与 wg，因此它跟整理同生共死（`Stop` 一并停掉它），不新开关闭链、 +// 不新开端口、不加 env 键 —— 组合根这边**零接线**。 // // 刻意**不做**的一件事（design §4.1 F）： // - 不把 sess-memorytidy 加进 flowPub 的图源白名单（默认关：整理是系统任务， @@ -54,11 +54,10 @@ func (a *app) wireMemoryTidy() error { if err != nil { return fmt.Errorf("memory tidy: %w", err) } - // 到期清扫（design/23 §19 的第 ② / ③ 步）挂到整理服务的生命周期上：它随心跳 - // 的 ctx 跑、随 a.memoryTidy.Stop 停（`design/23 §19.3-3` 的「可停」），因此 - // 这里**不新起一条关闭链**、不新开监听端口、不加任何 env 键。清扫自身的能力面 - // 只在 memory 包内（expire.go / expire_loop.go），组合根这一句只负责挂上去。 - svc.AttachExpirySweep() + // 到期清扫（design/23 §19 的第 ② / ③ 步）**不在组合根挂**：它起在 + // TidyService.Start 里，与整理心跳共用同一个 ctx 与 wg（见 expire_loop.go）。 + // 因此这里零改动即完成接线 —— 不新起一条关闭链、不新开监听端口、不加任何 + // env 键，`wire_services.go` 的 a.memoryTidy.Stop 会连同它一起停掉。 a.memoryTidy = svc return nil } diff --git a/cmd/cyrene/wire_memory_tidy_test.go b/cmd/cyrene/wire_memory_tidy_test.go index 2b7d957..e904030 100644 --- a/cmd/cyrene/wire_memory_tidy_test.go +++ b/cmd/cyrene/wire_memory_tidy_test.go @@ -69,6 +69,18 @@ func TestMemoryTidyIsWired(t *testing.T) { "main.go": { "a.memoryTidy.Start(ctx)", }, + // 到期清扫（切片 M3，design/23 §19 的第 ② / ③ 步）：临时记忆到期后要 + // **真删掉**，而它起在 TidyService.Start 里（与心跳同一个 cctx / wg）。 + // 少了这一句，过期行只会被读侧时效闸「挡在门外」而**永远不删** —— + // 不报错、不红任何别的用例，只是库里的临时层无界留存。因此这里把它 + // 钉成 seam：`TestTidyServiceStart_清扫随服务起停` 是它的运行时对照。 + "../../internal/runtime/memory/tidy_service.go": { + "go s.expirySweepLoop(cctx)", + }, + "../../internal/runtime/memory/expire_loop.go": { + "func (s *TidyService) expirySweepLoop(ctx context.Context)", + "DefaultExpireSweepInterval", + }, "wire_services.go": { // 关库之前先停整理（它每轮要用连接池）。 "a.memoryTidy.Stop(ctx)", diff --git a/internal/runtime/memory/expire_loop.go b/internal/runtime/memory/expire_loop.go index 77ca177..e44405b 100644 --- a/internal/runtime/memory/expire_loop.go +++ b/internal/runtime/memory/expire_loop.go @@ -2,7 +2,6 @@ package memory import ( "context" - "sync" "time" ) @@ -13,51 +12,17 @@ import ( // 审查同一单写者纪律，**不新开并发写口**、不新开监听端口、不加任何 `.env` 键 // （`.env` 已冻结，`design/13 §2.1`），也不碰会话 / 历史读数。 // -// 🔴 **挂在哪个 ctx 上**：清扫与整理**共用同一条生命周期** —— 它的 ctx 就是整理 -// 心跳的 ctx，它的 goroutine 登记进 TidyService 的 `wg`。因此 -// `cmd/cyrene/wire_services.go` 关闭链里的 `a.memoryTidy.Stop(ctx)` 会一并取消它 -// 并等它退出（§19.3-3 的「可停」），不必再为清扫加一条关闭路径、也不必改 -// `tidy_service.go`（那个文件归别的切片）。 +// 🔴 **挂点**：清扫与整理**同生共死** —— 它的 ctx 是 TidyService.Start 里建的那个 +// cctx（就是 `go s.loop(cctx)` 用的同一个），它的 goroutine 登记进同一个 `s.wg`。 +// 因此 `cmd/cyrene/wire_services.go` 关闭链里的 `a.memoryTidy.Stop(ctx)` 会一并 +// 取消它并等它退出（§19.3-3 的「可停」），不必再为清扫加一条关闭路径。 // -// 挂点用的是 TidyService 已有的注入缝 `wait`（心跳的等待原语，见 -// tidy_service.go:64-66）：它由心跳 goroutine 以**心跳的 cctx** 调用，因此「第一次 -// wait 被调用」既是「Start 已经跑起来、ctx 已经建好」的确定信号，又恰好把清扫要的 -// 那个可取消 ctx 递到手上。清扫随后跑在**自己的** goroutine 与**自己的**小时级 -// ticker 上，不借用整理的调度节奏（整理的周期是 24h 档，不能拿来当清扫周期）。 - -// AttachExpirySweep 把到期清扫挂到整理服务的生命周期上。它由组合根在装配期调用 -// （`cmd/cyrene/wire_memory_tidy.go`），每个服务实例调用一次。 -// -// 它在**第一次心跳等待**时启动清扫循环：此时 `Start` 已经建好可取消的 ctx， -// 而那个 ctx 就是心跳 goroutine 手里这一个。ctx 每次变化（Start → Stop → Start） -// 都会重新挂一个循环，因此「重启一次就永久失去清扫」不会发生。 -// -// 它不启动任何 goroutine 直到那一次等待发生：整理被配置关掉（`Start` 直接返回） -// 时，清扫也不会跑 —— 这与 `preinsert_gc.go` 的清扫随审查服务启停是同一条口径 -// （后台件随它的宿主启停，而不是凭空常驻）。 -func (s *TidyService) AttachExpirySweep() { - if s == nil { - return - } - inner := s.wait - if inner == nil { - inner = waitTidyTimer - } - var ( - mu sync.Mutex - attached context.Context - ) - s.wait = func(ctx context.Context, d time.Duration) bool { - mu.Lock() - if attached != ctx { - attached = ctx - s.wg.Add(1) - go s.expirySweepLoop(ctx) - } - mu.Unlock() - return inner(ctx, d) - } -} +// 🔴 挂点**必须**是 `Start` 里那处显式的 `go s.expirySweepLoop(cctx)`，不许改挂到 +// 某个注入缝上（例如包一层 `s.wait` 这个「心跳等待原语」）。理由是**可证性**： +// 包注入缝意味着清扫的存活依赖于「`loop` 每一轮都调 `s.wait`」这个**没有任何契约 +// 固定**的假设 —— 将来谁换了等待原语、或把循环改成 `select` + 自己的 timer，清扫 +// 会**静默消失**（不报错、不红测试，只是临时期限再也删不掉）。起在 `Start` 里， +// 读者一眼就能看到两个后台件同生共死。 // expirySweepLoop 是清扫的常驻循环：启动期先扫一次（清掉上次退出后到期的行）， // 此后按 DefaultExpireSweepInterval 扫一次。它随 ctx 取消退出（Stop 会 join 它）。 diff --git a/internal/runtime/memory/expire_loop_test.go b/internal/runtime/memory/expire_loop_test.go index 7b6b4a9..53ed47c 100644 --- a/internal/runtime/memory/expire_loop_test.go +++ b/internal/runtime/memory/expire_loop_test.go @@ -12,9 +12,11 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" ) -// 本文件锁定期清扫的**接线与后台形态**（design/23 §19）：挂点（AttachExpirySweep）、 -// 随 Stop 可停、一轮只写一行汇总日志，以及「删行时同步清掉向量」这条 finding 的 -// 集成断言。清扫自身的边界（哪些行删、哪些行留、审计长什么样）在 expire_test.go。 +// 本文件锁定期清扫的**接线与后台形态**（design/23 §19）：挂点（TidyService.Start +// 里的 `go s.expirySweepLoop(cctx)`）、随 Stop 可停、一轮只写一行汇总日志，以及 +// 「删行时同步清掉向量」这条 finding 的集成断言。清扫自身的边界（哪些行删、 +// 哪些行留、审计长什么样）在 expire_test.go。 +// // TestExpirySweepOnce_一轮只写一行汇总：逐条的那一行进审计，日志里一轮只有一行 // （§19.4「一轮汇总只进日志，绝不逐条把正文刷进日志」）。 func TestExpirySweepOnce_一轮只写一行汇总(t *testing.T) { @@ -60,18 +62,24 @@ func TestExpirySweepOnce_一轮只写一行汇总(t *testing.T) { } } -// TestAttachExpirySweep_挂上即扫_随Stop可停 钉住**接线**：清扫挂到整理服务的 -// 心跳 ctx 上（AttachExpirySweep），第一次心跳等待就启动、启动即扫一轮；ctx 取消 -// 后它的 goroutine 必须退出 —— `Stop` 走的就是「cancel + wg.Wait」，这里用同一个 -// 手法验证「可停」（§19.3-3）。 -func TestAttachExpirySweep_挂上即扫_随Stop可停(t *testing.T) { +// TestTidyServiceStart_清扫随服务起停 钉住**接线**（design/23 §19.3-3 的「可停」）： +// 清扫起在 TidyService.Start 里，与心跳共用同一个 cctx 与 wg，因此 +// +// ① Start 一跑它就立刻扫一轮（启动补扫，清掉上次退出后到期的行）； +// ② Stop 会取消它并等它退出 —— Stop 没超时即证明它真的随 ctx 走了。 +// +// 它同时是 `cmd/cyrene/wire_memory_tidy_test.go` 那条 seam 断言的运行时对照：把 +// Start 里那行 `go s.expirySweepLoop(cctx)` 摘掉，这条用例会红（行永远不会被删， +// 且 Stop 会因为 wg 计数对不上而超时）。 +func TestTidyServiceStart_清扫随服务起停(t *testing.T) { t.Parallel() ctx := context.Background() store := newTestStore(t) + // 10:00Z：不在安静窗（23–7）内 ⇒ Start 里的补跑判定为「不该跑」，整理不搅局。 now := tidyTestTime(t) row := addItem(t, store, MemoryItem{ PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship, - Content: "在心跳上被扫掉的到期行", ExpiresAt: now.Add(-time.Minute), + Content: "随服务启动被扫掉的到期行", ExpiresAt: now.Add(-time.Minute), }) var ( @@ -79,33 +87,26 @@ func TestAttachExpirySweep_挂上即扫_随Stop可停(t *testing.T) { audits []string ) svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), nil, fixedTidyClock(now), - func(who, what, result string) { + func(_, what, result string) { mu.Lock() defer mu.Unlock() audits = append(audits, what+" "+result) }) - // 假等待：不真等（真等会让用例挂 24h）。它只用来模拟「心跳调用了一次 wait」。 - svc.wait = func(context.Context, time.Duration) bool { return true } - svc.AttachExpirySweep() - - runCtx, cancel := context.WithCancel(ctx) - defer cancel() - if !svc.wait(runCtx, time.Hour) { - t.Fatal("包装后的 wait 必须是透传的（假等待返回 true，包装必须照原样返回）") + // 假等待：把整理心跳停在 ctx 上（真定时器要等 24h 档）。它只影响心跳 —— + // 清扫是**另一个** goroutine + 自己的小时级 ticker。 + svc.wait = func(ctx context.Context, _ time.Duration) bool { + <-ctx.Done() + return false } + svc.Start(ctx) + waitUntilRowGone(t, store, row.ID) - cancel() - done := make(chan struct{}) - go func() { - svc.wg.Wait() - close(done) - }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("清扫 goroutine 没有随 ctx 取消退出 —— 「可停」不成立（Stop 会卡在 wg.Wait）") + stopCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := svc.Stop(stopCtx); err != nil { + t.Fatalf("Stop 未能在预算内合上 —— 清扫没有随 ctx 退出 / 没登记进同一个 wg: %v", err) } mu.Lock() diff --git a/internal/runtime/memory/tidy_service.go b/internal/runtime/memory/tidy_service.go index 5875c7f..4aa7947 100644 --- a/internal/runtime/memory/tidy_service.go +++ b/internal/runtime/memory/tidy_service.go @@ -111,6 +111,11 @@ func (s *TidyService) Enabled() bool { return s.cfg.Enabled } // Start 启动心跳。它是幂等的：第二次调用是空操作。配置关闭时只记一条日志 // 并立即返回（关闭是显式选择，绝不静默）。ctx（可为 nil）是所有内部工作的 // 父级：取消它会像 Stop 一样停止服务。 +// +// 它**同时**起临时期限的到期清扫（design/23 §19 的第 ② / ③ 步，见 +// expire_loop.go）：清扫与心跳共用同一个 cctx 与 wg，因此它跟整理同生共死 —— +// 整理被配置关掉时清扫也不跑（与 `preinsert_gc.go` 的清扫随审查服务启停同口径）， +// `Stop` 会一并取消并 join 它。 func (s *TidyService) Start(ctx context.Context) { if !s.cfg.Enabled { s.log.Info("memory tidy disabled by configuration", "env", "MEMORY_TIDY_ENABLED") @@ -128,12 +133,22 @@ func (s *TidyService) Start(ctx context.Context) { cctx, cancel := context.WithCancel(ctx) s.cancel = cancel s.wg.Add(1) + // 到期清扫（design/23 §19 的第 ② / ③ 步）与心跳**同生共死**：同一个 cctx、 + // 同一个 wg，因此 Stop 会一并取消并 join 它（§19.3-3 的「可停」）。计数在 + // 这把锁里加（与上面那一笔同处）是刻意的：`Stop` 也持这把锁，这样 Add 与 + // Stop 的 `wg.Wait` 不可能撞在一起。 + s.wg.Add(1) s.mu.Unlock() // 持久事实先读进来：判定「要不要补跑」只看它，不看进程内状态。 s.loadPersistedLastRun(ctx) + // 两个后台件在这里一起起：整理心跳，以及临时期限的到期清扫 + // （expire_loop.go）。清扫跑在**自己的** goroutine 与**自己的**小时级 + // ticker 上 —— 整理的周期是 24h 档（intervalSec），不能拿来当清扫周期 + // （§19.1「到期后 ≤ 一个清扫周期，默认每小时」）。 go s.loop(cctx) + go s.expirySweepLoop(cctx) s.log.Info("memory tidy started", "interval", s.cfg.Interval.String(),