AskaEth
ed5005307a
- selfwake.go：WakeNow + 待跑槽容量 1（先探槽合并不过闸、槽空才过闸、同锁）+ 闸 4 轮/小时
（$DATA_DIR/config/backthink.json，零新 env 键）+ 审计四条 what + 启动日志两键 + 组合根接线
- G-A（主人已批）：TriggerUnknown 作零值、既有两值编号后移（语义不动），未知触发走**最紧分支**
- 5 个变异自检（拆闸/≤0 未回落/过闸挪到探槽前=G1 失效形态/槽无界/零值挪走）
⚠️ 服务侧已就绪但**她仍调不到**（自排工具面属 C1-3）⇒ 不得写「C 已完成」
307 lines
13 KiB
Go
307 lines
13 KiB
Go
package backthink
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"sync"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
)
|
||
|
||
// 本文件是**自我唤醒（自排一次思考）的本体**（design/12 §N C，切片 C1-2）。
|
||
//
|
||
// 形状（C.0）：她自己发起一次唤醒请求 → 先探**自排专用、容量 1** 的待跑槽
|
||
// （满 ⇒ 合并，**不过闸、不扣额度**）→ **槽空才**过窗口闸 → 单 worker 跑一轮
|
||
// TriggerSelfWake 思考。
|
||
//
|
||
// 三条不许破坏的约束（各有探针，见 C.8）：
|
||
// 1. **顺序**：探槽 → 过闸 → 入槽，三步在**同一把锁**内（C.8 判据 13）。颠倒成
|
||
// 「先过闸、再探槽」，被合并的请求也会白扣一次额度，与「合并不消耗预算」
|
||
// 自相矛盾（session.RateLimiter.Allow() **成功即消耗**）。
|
||
// 2. **闸前 normalize**（C.8 判据 14）：进闸的永远是已回落默认的值，因为
|
||
// RateLimiter 在 limit<=0 时 Allow() 恒 true 且不记录 ⇒ 未回落的值会把
|
||
// 「≤0 回落默认」静默变成「无上限」。
|
||
// 3. **容量 1 只限自排轮之间**：至多 1 轮自排在跑 + 至多 1 轮自排在排队；
|
||
// 心跳轮与事件轮照旧可与自排轮并发（C1 **不**引入全局回合互斥，G2）。
|
||
//
|
||
// 自排只保证「她能被叫醒思考」；要不要发出去仍由既有的 decide 级联 + PushGate 决定
|
||
// （C.4/C.5/C.7），本文件不碰推送策略。
|
||
|
||
// 自排的哨兵错误。用 errors.Is 判定；它们被直接返回或用 %w 包装。
|
||
var (
|
||
// ErrSelfWakeUnavailable 在服务未 Start 或已 Stop 时返回 —— **绝不假装排上**
|
||
// （红线：不伪造成功）。
|
||
ErrSelfWakeUnavailable = errors.New("backthink: self wake unavailable")
|
||
// ErrSelfWakeNotOwner 在非主人环境下请求自排时返回（工具面是第一道闸，
|
||
// 这里是第二道）。design/12 §C.3 / §C.5 的 denied。
|
||
ErrSelfWakeNotOwner = errors.New("backthink: self wake requires owner environment")
|
||
)
|
||
|
||
// SelfWakeReceipt 是一次自排请求的结果。字段与工具结果一一对应（design/12 §C.1/C.3），
|
||
// 因此「被上限拒绝」是她**看得见**的合法业务结果，而不是一个 error。
|
||
type SelfWakeReceipt struct {
|
||
// Accepted 报告是否真的排出一轮（合并 = true；超限 = false）。
|
||
Accepted bool
|
||
// Coalesced 报告这次请求与在队的那一轮合并了（不额外产生一轮、不花钱）。
|
||
Coalesced bool
|
||
// Reason 是拒绝/说明文案，可直接显示；成功时为空。
|
||
Reason string
|
||
// Used 是**本窗口已用**额度，在消耗之后取值（合并/拒绝路径取当前已用值，
|
||
// 不因这次请求 +1）。
|
||
Used int
|
||
// Limit 与 WindowSec 是本窗口的口径（已 normalize 的值），供面板与工具面显示。
|
||
Limit int
|
||
WindowSec int
|
||
}
|
||
|
||
// Auditor 是自排审计的窄接口：三字段与网关的 AuditEvent 同形（gateway/audit.go），
|
||
// 因此 *gateway.Server 直接满足它。审计出口是**晚绑定**的（见 sinks.go 的
|
||
// SetAuditSink）：网关比 backthink 晚组装。
|
||
type Auditor interface {
|
||
AuditEvent(who, what, result string)
|
||
}
|
||
|
||
// 自排审计行的 who 与四个动作。what 用点号小写的稳定标签（与 tool.reject.not_allowed
|
||
// 同族）；result 是可直接显示给主人的中文。
|
||
const (
|
||
selfWakeAuditWho = "backthink"
|
||
whatSelfWakeQueued = "backthink.self_wake.queued"
|
||
whatSelfWakeCoalesced = "backthink.self_wake.coalesced"
|
||
whatSelfWakeCapped = "backthink.self_wake.capped"
|
||
whatSelfWakeDenied = "backthink.self_wake.denied"
|
||
)
|
||
|
||
// WakeNow 是她自己排起一轮思考的入口（design/12 §N C.1）。它只负责**排**：
|
||
// 探槽、过闸、入槽，或如实返回合并/拒绝的收据；真正的思考由单 worker 跑。
|
||
//
|
||
// 步骤（顺序不许颠倒，C.1 步骤 1~4）：
|
||
// 1. 未 Start / 已 Stop ⇒ ErrSelfWakeUnavailable（不假装排上）；
|
||
// 2. 非主人环境 ⇒ ErrSelfWakeNotOwner + denied 审计（**不做事**）；
|
||
// 3. 先探容量 1 的待跑槽：满 ⇒ 合并（Accepted=true、Coalesced=true），
|
||
// **不过窗口闸、不扣额度**，在队的那一轮仍用先到的理由；
|
||
// 4. 槽空 ⇒ 过闸：拒 ⇒ Accepted=false + capped 审计 + 带 source 的 WARN
|
||
// （Allow() 返回 false 时本就未消耗 ⇒ 拒绝不占额度、不产生轮次）；
|
||
// 过 ⇒ 入槽（额度此刻已消耗）+ queued 审计。
|
||
//
|
||
// ctx 只用于身份环境与日志；待跑的那一轮不持有它（不引入延时/欠账语义，
|
||
// 也不让一次请求把调用方的 ctx 生命周期绑进 worker）。
|
||
func (b *Backthink) WakeNow(ctx context.Context, req SelfWakeRequest) (SelfWakeReceipt, error) {
|
||
gate, running := b.wakeGate()
|
||
if gate == nil || !running {
|
||
return SelfWakeReceipt{}, ErrSelfWakeUnavailable
|
||
}
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
// 第二道身份闸（第一道在工具面，C.3）：非主人环境一律拒绝，并留一条 denied
|
||
// 审计 —— 拒绝要看得见，不许静默。loop.SessionEnvFrom 对未挂载 env 的 ctx 返回
|
||
// 主人默认值（loop 的既有向后兼容口径），因此这条判定与工具面读的是同一个环境。
|
||
if env := loop.SessionEnvFrom(ctx); env.ActorEnv != restrict.ActorOwner {
|
||
b.auditEvent(whatSelfWakeDenied, fmt.Sprintf("非主人环境，拒绝自排（actor=%s）", env.ActorEnv))
|
||
return SelfWakeReceipt{}, ErrSelfWakeNotOwner
|
||
}
|
||
|
||
receipt, queued := gate.offer(req)
|
||
b.logSelfWake(req, receipt, queued)
|
||
return receipt, nil
|
||
}
|
||
|
||
// logSelfWake 把一次自排请求落到审计与日志上（四条 what 里的三条；denied 在
|
||
// WakeNow 的身份闸处记）。审计失败不改变结果：审计是留痕，不是门。
|
||
func (b *Backthink) logSelfWake(req SelfWakeRequest, receipt SelfWakeReceipt, queued bool) {
|
||
switch {
|
||
case receipt.Coalesced:
|
||
b.auditEvent(whatSelfWakeCoalesced,
|
||
fmt.Sprintf("已合并到排队中的那一轮（不额外占额度），理由：%s", req.Reason))
|
||
b.log.Info("backthink self wake coalesced",
|
||
"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
|
||
case receipt.Accepted:
|
||
b.auditEvent(whatSelfWakeQueued, fmt.Sprintf("已排入一轮自主思考，理由：%s", req.Reason))
|
||
b.log.Info("backthink self wake queued",
|
||
"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
|
||
case queued:
|
||
// 不可达：queued 为真时 Accepted 必为真。留着是为了让「排上了却没记审计」
|
||
// 在下一次改动里立刻显形，而不是静默少一条。
|
||
b.log.Warn("backthink self wake queued without audit", "reason", req.Reason)
|
||
default:
|
||
b.auditEvent(whatSelfWakeCapped, receipt.Reason)
|
||
b.log.Warn("backthink self wake capped",
|
||
"reason", req.Reason, "used", receipt.Used, "limit", receipt.Limit,
|
||
"window", b.cfg.SelfWakeWindow.String())
|
||
}
|
||
}
|
||
|
||
// auditEvent 把一条审计写进晚绑定的出口（未接线时静默跳过：审计缺失不阻塞业务）。
|
||
func (b *Backthink) auditEvent(what, result string) {
|
||
if a := b.currentAuditor(); a != nil {
|
||
a.AuditEvent(selfWakeAuditWho, what, result)
|
||
}
|
||
}
|
||
|
||
// wakeGate 在生命周期锁下取出自排闸并报告服务是否在跑。未 Start / 已 Stop 时
|
||
// running 为 false ⇒ 调用方返回 ErrSelfWakeUnavailable（不假装排上）。
|
||
func (b *Backthink) wakeGate() (*selfWakeGate, bool) {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
return b.selfWake, b.running
|
||
}
|
||
|
||
// selfWakeGate 是自排的**成本闸 + 待跑槽**（design/12 §N C.1/C.5）。它把三件事
|
||
// 锁在一起：容量 1 的待跑槽、每窗口上限的滑窗，以及「探槽 → 过闸 → 入槽」的强制顺序。
|
||
//
|
||
// 锁序：本闸的互斥量只包住自己的状态；RateLimiter 自带内部锁且绝不回调本闸，
|
||
// 因此不存在反向加锁路径（C.1 步骤 4 提到的「无锁序问题」）。
|
||
type selfWakeGate struct {
|
||
mu sync.Mutex
|
||
// limiter 是每窗口上限的滑窗（复用既有件 ⇒ 零新依赖、零新机制）。
|
||
// 它拿到手的必须是 normalize 之后的值：limit<=0 时它恒放行且不记录。
|
||
limiter *session.RateLimiter
|
||
limit int
|
||
window time.Duration
|
||
// granted 是「本窗口已用」的镜像，**只为上报**（收据的 Used 与审计文案）。
|
||
// 判定始终由 limiter 做；两者用同一个时间源与同一个窗口，因此不会各说各话。
|
||
granted []time.Time
|
||
// pending 是容量 1 的待跑槽。非 nil ⇒ 满了，后来的请求合并进它。
|
||
pending *SelfWakeRequest
|
||
// ready 只用来唤醒 worker（容量 1，非阻塞发送）。它不承载状态：
|
||
// 状态在 pending 上，因此不会出现「收到通知但槽是空的」以外的失配。
|
||
ready chan struct{}
|
||
}
|
||
|
||
// newSelfWakeGate 构建闸。limit/window 必须是**已 normalize** 的值
|
||
// （≤0 会得到恒放行的滑窗，见 cfg.go 的 normalize 注释）。
|
||
func newSelfWakeGate(limit int, window time.Duration) *selfWakeGate {
|
||
return &selfWakeGate{
|
||
limiter: session.NewRateLimiter(limit, window),
|
||
limit: limit,
|
||
window: window,
|
||
ready: make(chan struct{}, 1),
|
||
}
|
||
}
|
||
|
||
// offer 落实 C.1 步骤 2~4：**先探槽**（满 ⇒ 合并，不过闸、不扣额度）→ **槽空才过闸**
|
||
// → 入槽；三步在同一把锁内完成，因此两个并发自排不会同时看到「槽空」而各排一轮。
|
||
//
|
||
// 返回值 queued 报告是否真排出了一轮（合并 = false）。
|
||
func (g *selfWakeGate) offer(req SelfWakeRequest) (SelfWakeReceipt, bool) {
|
||
g.mu.Lock()
|
||
defer g.mu.Unlock()
|
||
|
||
if g.pending != nil {
|
||
// 合并路径**一次都不调用** Allow() —— 它成功即消耗，碰一下就是白扣额度。
|
||
return g.receiptLocked(true, true, ""), false
|
||
}
|
||
if !g.limiter.Allow() {
|
||
used := g.usedLocked()
|
||
receipt := SelfWakeReceipt{
|
||
Used: used,
|
||
Limit: g.limit,
|
||
WindowSec: int(g.window / time.Second),
|
||
Reason: fmt.Sprintf("已达自我唤醒上限（本窗口 %d/%d，窗口 %s），本轮未排队",
|
||
used, g.limit, g.window),
|
||
}
|
||
return receipt, false
|
||
}
|
||
|
||
g.granted = append(g.granted, time.Now())
|
||
queued := req
|
||
g.pending = &queued
|
||
select {
|
||
case g.ready <- struct{}{}:
|
||
default: // 已有一条通知在等：worker 取完槽后会再进内层循环，不会漏。
|
||
}
|
||
return g.receiptLocked(true, false, ""), true
|
||
}
|
||
|
||
// take 取走待跑槽（worker 用）。槽空 ⇒ ok=false，调用方回到等待。
|
||
func (g *selfWakeGate) take() (SelfWakeRequest, bool) {
|
||
g.mu.Lock()
|
||
defer g.mu.Unlock()
|
||
if g.pending == nil {
|
||
return SelfWakeRequest{}, false
|
||
}
|
||
req := *g.pending
|
||
g.pending = nil
|
||
return req, true
|
||
}
|
||
|
||
// pendingDepth 报告待跑槽是否占用（探针用：判据 13 要确认合并、14 要确认槽已空）。
|
||
func (g *selfWakeGate) pendingDepth() int {
|
||
g.mu.Lock()
|
||
defer g.mu.Unlock()
|
||
if g.pending == nil {
|
||
return 0
|
||
}
|
||
return 1
|
||
}
|
||
|
||
// receiptLocked 组装一张收据（调用方必须持有 g.mu）。Used 在**消耗之后**取值。
|
||
func (g *selfWakeGate) receiptLocked(accepted, coalesced bool, reason string) SelfWakeReceipt {
|
||
return SelfWakeReceipt{
|
||
Accepted: accepted,
|
||
Coalesced: coalesced,
|
||
Reason: reason,
|
||
Used: g.usedLocked(),
|
||
Limit: g.limit,
|
||
WindowSec: int(g.window / time.Second),
|
||
}
|
||
}
|
||
|
||
// usedLocked 报告本窗口已用额度并顺手丢弃过期的记录（调用方必须持有 g.mu）。
|
||
//
|
||
// 它是**只读口径的镜像**：与 limiter 用同一个时间源（time.Now）与同一个窗口，
|
||
// 因此报了「已用几档」，而不参与任何判定。
|
||
func (g *selfWakeGate) usedLocked() int {
|
||
cutoff := time.Now().Add(-g.window)
|
||
keep := 0
|
||
for _, t := range g.granted {
|
||
if t.After(cutoff) {
|
||
g.granted[keep] = t
|
||
keep++
|
||
}
|
||
}
|
||
g.granted = g.granted[:keep]
|
||
return keep
|
||
}
|
||
|
||
// selfWakeLoop 是自排 worker（design/12 §N C.1）：**单 goroutine** ⇒ 至多一轮自排在跑；
|
||
// 待跑槽容量 1 ⇒ 至多一轮自排在排队。两者合起来就是「排一次只跑一轮 + 最多额外欠 1 轮」。
|
||
//
|
||
// 它不做延时：槽里的请求立刻跑（重启不产生补跑风暴）。ctx 取消（Stop）时退出；
|
||
// 槽里若还留着一轮未跑，就地丢弃——没有持久化，也就没有欠账。
|
||
func (b *Backthink) selfWakeLoop(ctx context.Context) {
|
||
defer b.wg.Done()
|
||
gate := b.selfWake
|
||
if gate == nil {
|
||
return
|
||
}
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-gate.ready:
|
||
}
|
||
// 内层循环把当时可取的槽位全部取走：合并语义保证它至多取到一轮，
|
||
// 但循环写法让「取到空槽」成为正常出口，而不是靠时序假设。
|
||
for {
|
||
req, ok := gate.take()
|
||
if !ok {
|
||
break
|
||
}
|
||
b.runSelfWake(ctx, req)
|
||
}
|
||
}
|
||
}
|
||
|
||
// runSelfWake 跑一轮自排思考：折算走与事件同一条链（selfWakeTrigger → foldDeclared →
|
||
// escalateImport），随后完全复用既有的 think（记录、判定、推送、落库）。
|
||
func (b *Backthink) runSelfWake(ctx context.Context, req SelfWakeRequest) {
|
||
b.log.Info("backthink self wake round",
|
||
"reason", req.Reason, "declared", req.Declared.String())
|
||
b.think(ctx, selfWakeTrigger(req))
|
||
}
|