From 4c9cabce1be6244099491358b680b13b9eec34ea Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 21:26:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(channelbridge):=20=E7=AA=97=E5=8F=A3?= =?UTF-8?q?=E5=87=86=E5=85=A5=E8=90=BD=E5=88=B0=E8=BD=A6=E9=81=93=E5=A4=B4?= =?UTF-8?q?=20+=20lane.go=20=E6=8C=89=E8=81=8C=E8=B4=A3=E6=8B=86=E5=88=86?= =?UTF-8?q?=EF=BC=88M23=20=E5=A4=8D=E5=AE=A1=20r1=20P2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2-1(行为缺陷):窗口准入此前只在 prepareInbound 里算 `solo`,而 prepare 发生在 **窗口等待之后** —— 平台帧(action / 系统通知 / 我们自己的 message_sent 回执)照样 进窗、照样把滑动计时重置:一条回执就能把这一批的静默期一直续住(群友说了三句话, 她因为「撤回事件」迟迟不开口)。这与任务书「窗口只收 chat/image/voice/file」相反。 - `laneWindowAdmits(msg)` 成为**唯一准入判据**,只看结构化事实(ContentType + isSystemNoticeMessage,绝不解析正文); - `splitWindowItems` 在窗口等待阶段把到达的条目分成「进窗口」与「不进窗口」两半; - 非准入帧**不建窗口、不重置计时**(首条即到达即发车),窗口中途到达的只收进 solos, 这一批发车后**原样回插队首**各自单跑(保序、不合并、不丢弃); - prepareInbound 里的 `solo` 保留为第二道保险(适配器将来把平台事件标成 text 时, 仍会在组批前被切出去)。 P2-2(硬约束):lane.go 682 行 → 按职责纯搬移成 5 个文件(最大 246 行): lane.go(车道类型与注册表)/ lane_queue.go(入队与取队列)/ lane_worker.go(单消费者与回合槽)/ lane_drop.go(丢弃与停机诚实)/ lane_owner.go(主人私聊判据与车道键)。语义零改动:去注释后的声明集合 (34 个顶层声明,含函数体/字段/常量值)逐字节全等;既有用例一条未改。 顺带:把两处「IsOwner() && !IsGroupChat()」的注释改成代码真正用的判据 (ownerPrivateLane = !msg.IsGroup() && actor.IsOwner()),并记下这处 P3 分叉。 验证:全树 build/vet/test 全绿(73 包 ok);三包 -race 全绿; 新增 4 条准入用例在「准入失效」变异下三条必红、对照组保持绿。 --- internal/services/channelbridge/coalesce.go | 83 ++- .../channelbridge/coalesce_admission_test.go | 226 ++++++++ .../services/channelbridge/coalesce_lane.go | 22 +- .../channelbridge/coalesce_prepare.go | 12 +- internal/services/channelbridge/lane.go | 499 +----------------- internal/services/channelbridge/lane_drop.go | 131 +++++ internal/services/channelbridge/lane_owner.go | 68 +++ internal/services/channelbridge/lane_queue.go | 246 +++++++++ .../services/channelbridge/lane_worker.go | 108 ++++ internal/services/gateway/admin_coalesce.go | 5 +- 10 files changed, 873 insertions(+), 527 deletions(-) create mode 100644 internal/services/channelbridge/coalesce_admission_test.go create mode 100644 internal/services/channelbridge/lane_drop.go create mode 100644 internal/services/channelbridge/lane_owner.go create mode 100644 internal/services/channelbridge/lane_queue.go create mode 100644 internal/services/channelbridge/lane_worker.go diff --git a/internal/services/channelbridge/coalesce.go b/internal/services/channelbridge/coalesce.go index e31f070..54b4ddb 100644 --- a/internal/services/channelbridge/coalesce.go +++ b/internal/services/channelbridge/coalesce.go @@ -28,6 +28,7 @@ import ( "sync/atomic" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" ) @@ -184,18 +185,59 @@ func (b *Bridge) coalesceWindowFor(laneKey string, strangerPrivate bool) (coales return w, true } -// coalesceWait 在车道头部等一段静默期,把等待期间到达的消息**全部**并进这一批, -// 并返回发车原因。 +// laneWindowAdmits 是**窗口准入的唯一判据**(M23 任务书:「窗口只收 chat/image/voice/file; +// **不收**系统通知 / action 帧 / 我们自己出站回执」)。 +// +// 它只读消息自身的**结构化事实** —— ContentType 与「是不是平台帧」(isSystemNoticeMessage +// 同样只按 ContentType 判定),**绝不解析正文**、绝不猜语义。拒绝的两类理由: +// +// - **平台帧 / 系统通知 / 我们自己的 `message_sent` 回执**(ContentType=action):延迟 +// 2s 只为合并它们毫无收益,更糟的是**一条回执就能把窗口一直续住**(它们和群聊共用 +// 同一条车道键); +// - 其余非聊天类型:本来就没有「等她把话说完」的语义。 +// +// 不进窗口的条目**不是被丢弃**:它们走「到达即发车」的 solo 路径(见 coalesce_lane.go +// 的 runBatchLaneItem / coalesceWait 的 solos)。 +func laneWindowAdmits(msg adapter.UnifiedMessage) bool { + if isSystemNoticeMessage(msg) { + return false + } + switch msg.ContentType { + case adapter.ContentText, adapter.ContentImage, adapter.ContentVoice, adapter.ContentFile: + return true + default: + return false + } +} + +// splitWindowItems 把一片条目按**窗口准入**分成两半:进窗口的(保持原序)与不进窗口的 +// (同样保持原序)。两半都会被处理 —— 这是分类,不是剔除。 +func splitWindowItems(items []laneItem) (admitted, others []laneItem) { + for _, item := range items { + if laneWindowAdmits(item.msg) { + admitted = append(admitted, item) + continue + } + others = append(others, item) + } + return admitted, others +} + +// coalesceWait 在车道头部等一段静默期,把等待期间到达的**可进窗口**的消息并进这一批, +// 并返回(这一批、不进窗口的条目、发车原因)。 // // 语义(M23 蓝图 §4): // - **滑动**:新消息把计时重置到 w.window(「等她把话说完」); // - **绝对上限**:w.maxWait 锚定在批内**第一条**的到达时刻 —— 一个不停说话的群 -// 因此不会让计时永不归零(宁可晚回,不许不回)。 +// 因此不会让计时永不归零(宁可晚回,不许不回); +// - **准入**:只有 `laneWindowAdmits` 为真的条目参与攒批与重置计时;通知 / action 帧 / +// 出站回执只被**收进 solos**(原序),既不改计时也不进批 —— 否则「一条我们自己的 +// 回执把窗口续住」就是生产上的真实现场。 // // 它只做「等」与「收」:不判内容、不丢弃、不改顺序(收进来的顺序就是到达顺序: // 车道 FIFO ⇒ takeAll 的切片本身有序)。 -func (b *Bridge) coalesceWait(ctx context.Context, l *lane, held []laneItem, w coalesceWindow) ([]laneItem, string) { - items := held +func (b *Bridge) coalesceWait(ctx context.Context, l *lane, held []laneItem, w coalesceWindow) (items, solos []laneItem, reason string) { + items = held anchor := items[0].arrivedAt lastArrival := items[len(items)-1].arrivedAt b.coalesceStats.windowsCreated.Add(1) @@ -204,11 +246,11 @@ func (b *Bridge) coalesceWait(ctx context.Context, l *lane, held []laneItem, w c // 绝对上限先判:到点必发车(哪怕静默窗口因为不停有新消息而一直没归零)。 maxLeft := w.maxWait - now.Sub(anchor) if maxLeft <= 0 { - return items, coalesceFlushMaxWait + return items, solos, coalesceFlushMaxWait } quietLeft := w.window - now.Sub(lastArrival) if quietLeft <= 0 { - return items, coalesceFlushQuiet + return items, solos, coalesceFlushQuiet } wait := quietLeft if maxLeft < wait { @@ -216,24 +258,31 @@ func (b *Bridge) coalesceWait(ctx context.Context, l *lane, held []laneItem, w c } select { case <-l.batcher.after(wait): - more := b.takeAll(l) - if len(more) == 0 { - // 没有新消息:到点即发车,原因由「哪条上限先到」决定。 + seen, others := splitWindowItems(b.takeAll(l)) + solos = append(solos, others...) + if len(seen) == 0 { + // 没有新的**聊天**消息:到点即发车,原因由「哪条上限先到」决定 + // (不进窗口的帧不推迟这个判断)。 if w.maxWait-b.now().Sub(anchor) <= w.window-b.now().Sub(lastArrival) { - return items, coalesceFlushMaxWait + return items, solos, coalesceFlushMaxWait } - return items, coalesceFlushQuiet + return items, solos, coalesceFlushQuiet } - items = append(items, more...) + items = append(items, seen...) lastArrival = items[len(items)-1].arrivedAt case item := <-l.ch: - items = append(items, item) - if more := b.takeAll(l); len(more) > 0 { - items = append(items, more...) + if !laneWindowAdmits(item.msg) { + // 不进窗口:只登记到 solos,**绝不重置计时**。 + solos = append(solos, item) + continue } + items = append(items, item) + seen, others := splitWindowItems(b.takeAll(l)) + items = append(items, seen...) + solos = append(solos, others...) lastArrival = items[len(items)-1].arrivedAt case <-ctx.Done(): - return items, coalesceFlushStop + return items, solos, coalesceFlushStop } } } diff --git a/internal/services/channelbridge/coalesce_admission_test.go b/internal/services/channelbridge/coalesce_admission_test.go new file mode 100644 index 0000000..4d0af21 --- /dev/null +++ b/internal/services/channelbridge/coalesce_admission_test.go @@ -0,0 +1,226 @@ +package channelbridge + +// coalesce_admission_test.go 钉住 M23 的**窗口准入条件**(复审 r1 的 P2-1): +// +// 窗口只收 chat/image/voice/file;**不收**系统通知 / action 帧 / 我们自己出站回执。 +// +// 为什么这条要用例钉死(而不是靠注释):准入判断必须发生在**窗口等待阶段**(车道头), +// 而不是等窗口到了、prepareInbound 跑完才发现「这条本来不该进批」。否则一条我们自己的 +// `message_sent` 回执(它与群聊共用同一条车道键)就能把这一批的静默计时**一直续住**: +// 群友说了三句话,她却因为「撤回事件」「入群事件」「我们自己的回执」而迟迟不开口。 +// +// 四条断言(每条都能被「准入失效」这一变异打红): +// - 窗口中途的 action 帧**不重置计时**(剩余等待仍是 500ms,发车时刻不变); +// - 系统通知**不进这一批**(批内 user 行只有聊天消息); +// - 单条出站回执**不建窗口**(定时器一次都没被请求); +// - 对照组:普通聊天消息**仍然**重置计时(否则上面三条可能只是「窗口根本不工作」)。 + +import ( + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// coalesceActionFrame 是一条**平台帧**(action):撤回 / 入群 / typing / 未知 post_type。 +// 它的生产形状与本群聊天共用车道的键(ChannelType=group + GroupID),ContentType=action。 +func coalesceActionFrame(platform, groupID, postType string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: platform, + ChannelID: groupID, + ChannelType: adapter.ChannelGroup, + GroupID: groupID, + SenderID: "9", + Sender: adapter.Actor{Role: adapter.RoleMember}, + ContentType: adapter.ContentAction, + RawData: []byte(`{"post_type":"` + postType + `"}`), + } +} + +// coalesceOutboundReceipt 是**我们自己那条消息的回执**(OneBot 的 message_sent): +// 它是一条 ContentType=action、没有正文的群帧 —— 也正是「能把窗口续住」的那条。 +func coalesceOutboundReceipt(platform, groupID string) adapter.UnifiedMessage { + msg := coalesceActionFrame(platform, groupID, "message_sent") + msg.SenderID = "100" // 她自己 + msg.Sender = adapter.Actor{Role: adapter.RoleOwner} + return msg +} + +// waitTimerCalls 等到定时器工厂至少被请求 n 次,返回到目前为止的**请求时长序列**。 +func waitTimerCalls(t *testing.T, f *coalesceBridge, n int) []time.Duration { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if d := f.timers.durations(); len(d) >= n { + return d + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timer factory called %d times, want at least %d", f.timers.callCount(), n) + return nil +} + +// fireLatest 按**最后一个**待定定时器自己的时长推进时钟并触发它。 +// +// 为什么是最后一个:窗口被重置/续臂时,旧的定时器通道留在那里(不会被触发), +// 真正决定发车时刻的是**最近一次**请求的那个。 +func (t *coalesceTimers) fireLatest(tb *testing.T, clock *laneClock) { + tb.Helper() + t.mu.Lock() + if len(t.pending) == 0 { + t.mu.Unlock() + tb.Fatalf("no pending coalesce timer to fire") + } + timer := t.pending[len(t.pending)-1] + t.pending = t.pending[:len(t.pending)-1] + t.mu.Unlock() + clock.advance(timer.d) + timer.ch <- clock.Now() +} + +// TestCoalesceActionFrameDoesNotResetWindow 断言**准入在窗口等待阶段就生效**: +// 一条平台帧(action)落在窗口中途,既不进批、也不重置静默计时 —— +// 剩余等待仍是 500ms,发车时刻仍是「m1 到达 + 2.0s」。 +// +// 变异:让 laneWindowAdmits 恒真(= 修前行为)⇒ 第二次请求的时长变成 2000ms,本用例红。 +func TestCoalesceActionFrameDoesNotResetWindow(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + f := newCoalesceBridge(t, Options{ + Loop: chat, + Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, + }) + f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) + // 第一条聊天消息 ⇒ 一个完整窗口(2.0s)。 + if d := waitTimerCalls(t, f, 1); d[0] != 2000*time.Millisecond { + t.Fatalf("first window = %s, want 2s", d[0]) + } + + // 窗口走到 1.5s 时来了一条平台帧。 + f.clock.advance(1500 * time.Millisecond) + f.send(coalesceActionFrame("telegram", "-20", "notice.group_recall")) + + // 🔴 关键断言(状态,不是结果):下一次请求的**剩余**时长仍是 500ms。 + // 若这条帧被当成聊天消息(修前),计时会重置成完整的 2000ms。 + d := waitTimerCalls(t, f, 2) + if d[1] != 500*time.Millisecond { + t.Fatalf("re-armed window = %s, want 500ms (an action frame must not reset the quiet window)", d[1]) + } + + f.timers.fireLatest(t, f.clock) + f.waitRuns(t, 1) + f.waitIdle(t) + + if n := chat.runCount(); n != 1 { + t.Fatalf("turns started = %d, want 1 (the frame must not add a turn)", n) + } + if got := userTextsIn(chat.runAt(0).messages); len(got) != 1 || got[0] != "m1" { + t.Fatalf("batch user messages = %v, want [m1] (the frame must not join the batch)", got) + } + stats := f.b.CoalesceStats() + if stats.WindowsCreated != 1 || stats.BatchesTotal != 1 { + t.Fatalf("stats = %+v, want one window and one batch", stats) + } +} + +// TestCoalesceSystemNoticeDoesNotJoinBatch 断言系统通知**不进这一批**: +// 两条聊天消息 + 中途一条通知 ⇒ 一轮、批内只有那两条聊天消息。 +func TestCoalesceSystemNoticeDoesNotJoinBatch(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + f := newCoalesceBridge(t, Options{ + Loop: chat, + Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, + }) + f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) + f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")) + waitTimerCalls(t, f, 1) + + f.clock.advance(1500 * time.Millisecond) + f.send(coalesceActionFrame("telegram", "-20", "notice.group_increase")) + // 与 action 帧同理:通知也不许重置计时(剩余仍是 500ms)。 + // 变异(准入失效)⇒ 这里拿到 2s,本用例红。 + if d := waitTimerCalls(t, f, 2); d[1] != 500*time.Millisecond { + t.Fatalf("re-armed window = %s, want 500ms (a system notice must not reset the quiet window)", d[1]) + } + + f.timers.fireLatest(t, f.clock) + f.waitRuns(t, 1) + f.waitIdle(t) + + if n := chat.runCount(); n != 1 { + t.Fatalf("turns started = %d, want 1 (the notice must not drive its own turn here)", n) + } + if got := userTextsIn(chat.runAt(0).messages); len(got) != 2 || got[0] != "m1" || got[1] != "m2" { + t.Fatalf("batch user messages = %v, want [m1 m2] only", got) + } + if stats := f.b.CoalesceStats(); stats.BatchesTotal != 1 { + t.Fatalf("batches = %d, want 1 (the notice is not a batch)", stats.BatchesTotal) + } +} + +// TestCoalesceOutboundReceiptCreatesNoWindow 断言**我们自己的出站回执**不建窗口: +// 单条回执到达 ⇒ 定时器工厂一次都没被请求、没有窗口、没有批、也没有回合。 +// +// 变异:让 laneWindowAdmits 恒真 ⇒ 回执会建一个窗口(定时器被请求、windows_created=1), +// 本用例红 —— 这条正是「一条回执就能把窗口续住」的最小复现。 +func TestCoalesceOutboundReceiptCreatesNoWindow(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + f := newCoalesceBridge(t, Options{ + Loop: chat, + Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, + }) + f.send(coalesceOutboundReceipt("telegram", "-20")) + // 先看「有没有窗口」再等空闲:修前会建一个永不触发的窗口(calls=1), + // 这里因此留下的是「窗口被建出来了」这条最直接的证据,而不是一句「车道没退休」。 + // 轮询到车道退休(或 300ms)是必要的:投递返回时 worker 可能还没起势, + // 一次性读 callCount() 会让这条断言在修前也偶尔通过。 + deadline := time.Now().Add(300 * time.Millisecond) + for time.Now().Before(deadline) { + if calls := f.timers.callCount(); calls != 0 { + t.Fatalf("coalesce timer factory called %d times for an outbound receipt, want 0", calls) + } + if f.b.LaneStats().ActiveLanes == 0 { + break + } + time.Sleep(time.Millisecond) + } + stats := f.b.CoalesceStats() + if stats.WindowsCreated != 0 || stats.BatchesTotal != 0 { + t.Fatalf("stats = %+v, want no window and no batch for a lone receipt", stats) + } + if n := chat.runCount(); n != 0 { + t.Fatalf("turns started = %d for a lone receipt, want 0", n) + } + f.waitIdle(t) +} + +// TestCoalesceChatStillResetsWindow 是对照组:普通聊天消息**仍然**重置计时 +// (否则上面三条可能只是「窗口本来就不工作」这种伪结论)。 +func TestCoalesceChatStillResetsWindow(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + f := newCoalesceBridge(t, Options{ + Loop: chat, + Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, + }) + f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) + if d := waitTimerCalls(t, f, 1); d[0] != 2000*time.Millisecond { + t.Fatalf("first window = %s, want 2s", d[0]) + } + f.clock.advance(1500 * time.Millisecond) + f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")) + if d := waitTimerCalls(t, f, 2); d[1] != 2000*time.Millisecond { + t.Fatalf("re-armed window = %s, want a full 2s (a chat message resets the window)", d[1]) + } + f.timers.fireLatest(t, f.clock) + f.waitRuns(t, 1) + if got := userTextsIn(chat.runAt(0).messages); len(got) != 2 { + t.Fatalf("batch user messages = %v, want both chat messages", got) + } +} diff --git a/internal/services/channelbridge/coalesce_lane.go b/internal/services/channelbridge/coalesce_lane.go index cc5f64a..3708069 100644 --- a/internal/services/channelbridge/coalesce_lane.go +++ b/internal/services/channelbridge/coalesce_lane.go @@ -25,19 +25,29 @@ import ( // 这段期间攒下的整片取出 ⇒ 一轮续跑(见 flushLaneBatch 的 continued 计数)。 func (b *Bridge) runBatchLaneItem(ctx context.Context, l *lane, first laneItem) { w, windowed := b.coalesceWindowFor(l.key, laneStrangerPrivate(first)) - if !windowed { - // 关窗(或该会话不适用窗口):行为**逐字回到 M22** —— 到达即投递、一条一轮。 - // 刻意**不** takeAll:多取的条目会让「特性关闭」变成「悄悄合并」。 + if !windowed || !laneWindowAdmits(first.msg) { + // 三种「到达即发车」:关窗(逐字回到 M22)、该会话不适用窗口、以及**不进窗口的帧** + // (系统通知 / action 帧 / 我们自己的出站回执)。最后一种刻意**不**建窗口: + // 一条回执不该让这一批的静默计时重新开始(M23 任务书的窗口准入条件)。 + // 刻意**不** takeAll:多取的条目会让「不合并」变成「悄悄合并」。 b.flushLaneBatch(ctx, l, []laneItem{first}, w, false, coalesceFlushOff) return } - items := append([]laneItem{first}, b.takeAll(l)...) + items, solos := splitWindowItems(b.takeAll(l)) + items = append([]laneItem{first}, items...) var flushReason string - items, flushReason = b.coalesceWait(ctx, l, items, w) + var moreSolos []laneItem + items, moreSolos, flushReason = b.coalesceWait(ctx, l, items, w) + solos = append(solos, moreSolos...) if flushReason == coalesceFlushStop { - b.abandonAtStop(l, items) + b.abandonAtStop(l, append(append([]laneItem{}, items...), solos...)) return } + if len(solos) > 0 { + // 不进窗口的帧原样**插回队首**:本批先发车,随后它们各自单跑(一条一轮、 + // 不建窗口)。回插保序:它们比队列里其余条目都早,front 因此在 ch/spill 之前。 + b.requeueFront(l, solos) + } b.flushLaneBatch(ctx, l, items, w, true, flushReason) } diff --git a/internal/services/channelbridge/coalesce_prepare.go b/internal/services/channelbridge/coalesce_prepare.go index 3e9495c..6c15a7f 100644 --- a/internal/services/channelbridge/coalesce_prepare.go +++ b/internal/services/channelbridge/coalesce_prepare.go @@ -104,13 +104,13 @@ func (b *Bridge) prepareInbound(ctx context.Context, item laneItem) (preparedInb }, true } -// coalesceAdmitsMessage 报告这条入站是否允许进静默窗口(M23 蓝图 §11 第 4 条): +// coalesceAdmitsMessage 是窗口准入的**第二道保险**(第一道、也是真正生效的那道在车道头: +// coalesce.go 的 laneWindowAdmits —— 窗口等待阶段就按它分类,非准入条目根本进不了批)。 // -// exit 条件只有两类 —— **系统通知**(`notice.Applicable`,含升级成真会话的那些)与 -// **非聊天内容类型**(action 帧:typing/join/leave/recall 等)。 -// -// 理由不是省事:延迟 2s 只为合并它们毫无收益,还会污染留存时序;而「升级的真会话」 -// 与「群友聊天」混成一次回答,等于让平台事件替群友说话。它们因此走 solo(单独一轮)。 +// 它比车道头多知道一件事:`notice.Applicable`(那要 preEntryReview 跑完才有结论)。 +// 但 isSystemNoticeMessage 只按 ContentType=action 判定,而 action 帧在车道头已经被挡下, +// 所以这里实际上永远为真 —— 保留它是为了让「未来某个适配器把平台事件标成 text」这类 +// 变更仍然在组批前被 solo 切出去(runBatchSegments),而不是混进群友的同一轮回答。 func coalesceAdmitsMessage(msg adapter.UnifiedMessage, notice noticeOutcome) bool { if notice.Applicable { return false diff --git a/internal/services/channelbridge/lane.go b/internal/services/channelbridge/lane.go index 0f591fd..f247d8b 100644 --- a/internal/services/channelbridge/lane.go +++ b/internal/services/channelbridge/lane.go @@ -31,13 +31,11 @@ package channelbridge // 所以一条排队中的消息看到的是它开始那一刻的最新历史,而不是入队时的快照。 import ( - "context" "log/slog" "sync" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" - "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" ) @@ -124,8 +122,9 @@ type lane struct { // newDirectLane 构造**主人私聊**的 direct 车道:它没有任何窗口状态,也不持有定时器 // 工厂(不传 batchWindow ⇒ 拿不到)。 // -// 🔴 车道类型由 **actor 元数据**(`IsOwner() && !IsGroupChat()`)决定,配置无权翻转: -// 唯一能选 direct 的地方是 enqueueInbound 的 ownerPrivateLane 判定。主人私聊是 +// 🔴 车道类型由 **actor 元数据**决定(判据 `ownerPrivateLane`:`!msg.IsGroup() && +// actor.IsOwner()`,与 M22 的车道键/丢弃策略同源),配置无权翻转:唯一能选 direct +// 的地方是 enqueueInbound 的 ownerPrivateLane 判定。主人私聊是 // 「她和我说话」,2s 迟滞在私聊里没有任何收益(只有被拖慢的观感), // 且 M22 已裁定主人私聊永不丢弃、不受陈旧度闸约束 —— 本函数是同一取向的延续。 func newDirectLane(key string, ch chan laneItem, neverDrop bool) *lane { @@ -188,495 +187,3 @@ func newLaneSet(depth int, staleAfter time.Duration, after func(time.Duration) < lanes: make(map[string]*lane), } } - -// LaneStats 是车道的计数快照:**被丢弃的入站有多少、为什么**(不变量 ③ 的可审计面), -// 加上主人私聊溢出区的当前积压。 -type LaneStats struct { - // ActiveLanes 是此刻有排队或在跑的会话车道数。 - ActiveLanes int - // DroppedStale 是因陈旧度闸被丢弃的条目累计数(超龄不补答)。 - DroppedStale int64 - // DroppedOverflow 是车道满时**在门前被拒收**的条目累计数。它只可能落在群聊 / - // 陌生人私聊上:主人私聊车道永不拒收(见 offer 的 neverDrop)。 - DroppedOverflow int64 - // Backlog 是主人私聊溢出区里此刻还排着的条目数(其余车道恒为 0)。 - Backlog int -} - -// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。 -// -// 生产可见性的口径(复审 P2-3):**不新开 HTTP 端点**(那是契约变更,归塔)。真正 -// 生产可见的是每一条丢弃 WARN 里带的 `dropped_stale_total` / `dropped_overflow_total` -// (累计计数随日志落地)以及 `onRejected` 审计轨;本方法只是给进程内调用方(管理面板 -// 将来接线、运维探针、测试)读同一组数字的窄口,不构成对外契约。 -func (b *Bridge) LaneStats() LaneStats { - if b.lanes == nil { - return LaneStats{} - } - b.lanes.mu.Lock() - defer b.lanes.mu.Unlock() - backlog := 0 - for _, l := range b.lanes.lanes { - backlog += len(l.spill) - } - return LaneStats{ - ActiveLanes: len(b.lanes.lanes), - DroppedStale: b.lanes.droppedStale, - DroppedOverflow: b.lanes.droppedOverflow, - Backlog: backlog, - } -} - -// enqueueInbound 把一条**已过流控闸**的入站消息投进它所属会话的车道,并立刻返回 —— -// 这是「读与回水解耦」的全部:适配器的读循环调用它之后马上可以继续读 socket。 -// -// 🔴 这就是 M23 静默窗口合并的**入队 seam**:入参是一条**已过闸门的 prepared item** -// (Validate/ResolveActor/Admit 都已在本函数的调用方 onMessage 里做完,连同解析结果 -// actor 一起带进来)。车道头部只做「等(窗口)+ 发车」,准入判断不在这里。 -// -// 三件事在这里一次性做完: -// - 认领车道(没有就建一条并起 worker)—— **类型由 actor 元数据决定**: -// 主人私聊 ⇒ direct(无窗口),其余 ⇒ batch(有窗口); -// - 入队(群聊/陌生人满则**在门前拒收这一条**并留痕;主人私聊**永不拒收**); -// - 与 Stop 的 `wg.Wait` 对齐:`b.wg.Add` 与 `closed` 的判定在同一把 `b.mu` 下, -// 因此 Stop 要么看见这个工作项、要么我们看见 closed 而不入队,二者必居其一。 -// -// 锁顺序:全树只有这一处同时取两把锁,顺序恒为 `b.mu` → `lanes.mu`。 -func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, actor adapter.Actor) { - if b.lanes == nil { - return - } - item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()} - key := laneKey(msg, actor) - // 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**: - // 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。 - neverDrop := ownerPrivateLane(msg, actor) - spec := b.laneSpecFor(key, item, neverDrop) - - b.mu.Lock() - if b.closed { - b.mu.Unlock() - b.lanes.log.Debug("bridge closed, inbound not queued", - "platform", msg.Platform, "channel", msg.ChannelID) - return - } - l, start, rejected := b.lanes.offer(key, item, spec) - if start { - b.wg.Add(1) - } - b.mu.Unlock() - - if start { - go b.laneWorker(ctx, l) - } - if rejected { - b.dropInbound(item, laneDropReasonOverflow, "lane queue full") - } -} - -// laneSpecFor 决定一条车道的形状:**类型只由 actor 元数据决定** -// (`IsOwner() && !IsGroupChat()` ⇒ direct),窗口参数只影响容量。 -// -// 容量:batch 车道至少要能装下一整批,否则「一批 N 条」在结构上不可达 -// (队列深度 8 时第 9 条就会在门前被拒收,合并就成了摆设)。因此 -// capacity = max(默认深度, 生效的单批上限);direct 车道用默认深度。 -// 内存口径因此是「活跃会话数 × 批上限」,而批上限本身被 clamp 在 [1,50]。 -func (b *Bridge) laneSpecFor(key string, item laneItem, neverDrop bool) laneSpec { - spec := laneSpec{neverDrop: neverDrop, batch: !neverDrop, capacity: b.lanes.depth} - if neverDrop { - return spec - } - if w, ok := b.coalesceWindowFor(key, laneStrangerPrivate(item)); ok && w.maxBatch > spec.capacity { - spec.capacity = w.maxBatch - } - return spec -} - -// offer 在**一个临界区**内完成「查/建车道 + 入队」,使 `retireLaneIfIdle` 不可能夹在 -// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。 -// -// 队列满时的策略分两档(都不是静默的): -// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按倍率 -// (ownerLaneBacklogWarn 及其翻倍)留 WARN —— 一个白名单账号的积压只反映「她落后了」, -// 不构成第三方洪泛面; -// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里 -// 的条目因此一条都不会凭空消失 —— 窗口的 takeAll 依赖这一点;被拒收的那条 -// 由调用方留痕(WARN + 计数 + 审计)。 -// -// 🔴 **FIFO 的关键**:`spill` 一旦非空,后面的条目**一律继续追加到 spill**,不再回写 -// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车(复审实测顺序 -// `[m1 m2 m3 m5 m4]`)。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」: -// spill 里的一定晚于 ch 里的,而 spill 一旦开始就用到底(直到它被取空)。 -// -// 同理,`front`(回插队首区)里的一定**早于** ch 与 spill 里的(只有 worker 自己 -// 在发车前把超量的尾部放进去),所以读取顺序恒为 front → ch → spill。 -func (s *laneSet) offer(key string, item laneItem, spec laneSpec) (l *lane, start bool, rejected bool) { - s.mu.Lock() - defer s.mu.Unlock() - l = s.lanes[key] - if l == nil { - capacity := spec.capacity - if capacity <= 0 { - capacity = s.depth - } - ch := make(chan laneItem, capacity) - if spec.batch { - l = newBatchLane(key, ch, spec.neverDrop, &batchWindow{after: s.after}) - } else { - l = newDirectLane(key, ch, spec.neverDrop) - } - s.lanes[key] = l - start = true - } - if l.neverDrop && len(l.spill) > 0 { - s.spillLocked(l, item) - return l, start, false - } - select { - case l.ch <- item: - return l, start, false - default: - } - if l.neverDrop { - s.spillLocked(l, item) - return l, start, false - } - return l, start, true -} - -// spillLocked 把条目追加进主人私聊车道的溢出区,并按**倍率**告警 -// (16/32/64/128… 每翻一倍一条 WARN):固定水位只会在恰好等于时响一次,之后继续涨就 -// 再无信号,而「她落后了多少」正是主人侧唯一可观测的量。调用方必须持有 s.mu。 -func (s *laneSet) spillLocked(l *lane, item laneItem) { - l.spill = append(l.spill, item) - if n := len(l.spill); n >= ownerLaneBacklogWarn && isPowerOfTwo(n) { - s.log.Warn("owner private lane backlog is growing, still queueing", - "backlog", n, "lane", l.key, - "platform", item.msg.Platform, "channel", item.msg.ChannelID) - } -} - -// isPowerOfTwo 报告 n 是否为 2 的幂(n ≥ 1)。 -func isPowerOfTwo(n int) bool { return n > 0 && n&(n-1) == 0 } - -// ownerPrivateLane 报告这条消息是否属于「主人私聊」车道 —— 也就是**永不丢弃**的那条。 -// 判据与 laneKey 完全同源(私聊 + actor 元数据说他是主人),因此不会出现「键说是主人、 -// 丢弃策略说不是」的两套口径。这里绝不读正文:身份只来自 actor 元数据 + 白名单。 -func ownerPrivateLane(msg adapter.UnifiedMessage, actor adapter.Actor) bool { - return !msg.IsGroup() && actor.IsOwner() -} - -// nextLaneItem 按 FIFO 取下一条:先取回插队首区(front),再把缓冲区取完,最后取 -// 主人私聊的溢出区(三者的先后由 offer / requeueFront 的写入口径保证)。 -// ok=false 表示此刻真的没有待办。 -func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) { - b.lanes.mu.Lock() - if len(l.front) > 0 { - item := l.front[0] - l.front = l.front[1:] - b.lanes.mu.Unlock() - return item, true - } - b.lanes.mu.Unlock() - select { - case item := <-l.ch: - return item, true - default: - } - b.lanes.mu.Lock() - defer b.lanes.mu.Unlock() - if len(l.spill) == 0 { - return laneItem{}, false - } - item := l.spill[0] - copy(l.spill, l.spill[1:]) - l.spill = l.spill[:len(l.spill)-1] - return item, true -} - -// takeAll 把车道上**此刻**排着的条目整片取出(顺序 = 到达顺序),供静默窗口的 -// 批组装使用。它是合并阶段**唯一**的取队列入口(M23 蓝图 §4-1): -// -// - 只有车道的单消费者(该车道唯一的 worker)调用它 ⇒ 不存在两个消费者各抓一半; -// - 取出即「这批就是这些」,等待期间新到的继续落进队列,下一轮 takeAll 再取; -// - 它**不**做任何筛选:不丢、不重排、不改内容(窗口内零丢失的结构保证)。 -func (b *Bridge) takeAll(l *lane) []laneItem { - b.lanes.mu.Lock() - out := l.front - l.front = nil - b.lanes.mu.Unlock() - for { - select { - case item := <-l.ch: - out = append(out, item) - continue - default: - } - b.lanes.mu.Lock() - if len(l.spill) > 0 { - out = append(out, l.spill...) - l.spill = nil - } - b.lanes.mu.Unlock() - return out - } -} - -// requeueFront 把一批**尚未发车**的条目按原顺序插回队首,成为下一批。 -// -// 它是「单批上限不是丢弃」的落点(M23 蓝图 §4-3):超量的**尾部**原样回插、保序、 -// 在锁内、在发车之前。回插到 front 而不是写 ch:ch 有界且写满即拒收, -// 而 front 无界但只由 worker 在发车前写入,写入量与「刚拆出来的尾部」相等。 -func (b *Bridge) requeueFront(l *lane, items []laneItem) { - if len(items) == 0 { - return - } - b.lanes.mu.Lock() - merged := make([]laneItem, 0, len(items)+len(l.front)) - merged = append(merged, items...) - merged = append(merged, l.front...) - l.front = merged - b.lanes.mu.Unlock() -} - -// laneWorker 是某会话车道的唯一执行者:串行跑完每一条,队列空了就把车道交还注册表 -// (空闲车道不占内存,也不会把「这个人再说话」变慢)。 -// -// **回合中续跑(⑤)就落在这里**:不存在任何 "turn in flight" 标志 —— 「回合在跑」 -// 在结构上就等于「worker 正忙」,跑回合期间新消息落进同一个队列,回合结束回到循环头, -// takeAll 自然把这段期间攒下的消息整片取出 ⇒ **一轮续跑**(不是 N 轮,也不是 0 条)。 -func (b *Bridge) laneWorker(ctx context.Context, l *lane) { - defer b.wg.Done() - for { - // 桥已停:队列里还没发车的条目**如实登记**(绝不静默消失),然后退休。 - if ctx.Err() != nil { - b.abandonLaneAtStop(l) - b.retireLane(l) - return - } - if item, ok := b.nextLaneItem(l); ok { - b.dispatchLaneItem(ctx, l, item) - continue - } - // 队列空了才谈退休:摘表与入队用同一把锁,因此「摘下」与「刚有人塞进来」不会 - // 交错成条目丢失(见 retireLaneIfIdle 的说明)。 - if b.retireLaneIfIdle(l) { - return - } - select { - case <-ctx.Done(): - // 桥停了:把车道摘掉,未处理的条目随进程取消一起作废(Stop 的语义)。 - b.retireLane(l) - return - case item := <-l.ch: - b.dispatchLaneItem(ctx, l, item) - } - } -} - -// dispatchLaneItem 按**车道类型**分派:direct(主人私聊)一条一条跑;batch 走窗口 -// 引擎(见 coalesce_lane.go 的 runBatchLaneItem)。 -// -// 分派依据是 `l.batcher == nil` —— 一个**结构**判据,不是配置开关:direct 车道的 -// 类型里根本没有 batcher 字段值,窗口代码对它不可达。 -func (b *Bridge) dispatchLaneItem(ctx context.Context, l *lane, item laneItem) { - if l.batcher == nil { - b.runDirectLaneItem(ctx, l, item) - return - } - b.runBatchLaneItem(ctx, l, item) -} - -// runDirectLaneItem 跑 direct 车道(= 主人私聊)的一条:先取全局回合槽(**在线程池里 -// 等,不在读循环里等**),再过陈旧度闸,然后走逐条闸门 + 单条回合。**没有任何等待窗口**。 -func (b *Bridge) runDirectLaneItem(ctx context.Context, l *lane, item laneItem) { - if !b.acquireTurn(ctx) { - return - } - defer b.releaseTurn() - - waited := b.now().Sub(item.arrivedAt) - if b.staleInbound(item, waited) { - return - } - b.logLaneStart(item, waited, 1) - p, ok := b.prepareInbound(ctx, item) - if !ok { - l.lastTurnEnd = b.now() - return - } - b.runBatch(ctx, []preparedInbound{p}) - l.lastTurnEnd = b.now() -} - -// acquireTurn 取一个全局回合槽(maxConcurrent 个)。ctx 取消时返回 false:桥停了 -// 就不再开新回合(carried 条目由 Stop 的诚实登记负责,见 coalesce_lane.go)。 -func (b *Bridge) acquireTurn(ctx context.Context) bool { - select { - case b.sem <- struct{}{}: - return true - case <-ctx.Done(): - return false - } -} - -// releaseTurn 归还一个全局回合槽。 -func (b *Bridge) releaseTurn() { <-b.sem } - -// staleInbound 实现陈旧度闸:轮到时已超龄 ⇒ 丢弃不补答,并留 WARN + 计数 + 审计。 -// 返回 true 表示这条已被丢弃。waited 由调用方算好(含等全局回合槽的时间)。 -func (b *Bridge) staleInbound(item laneItem, waited time.Duration) bool { - if !laneStaleEligible(item) { - return false - } - if waited <= b.lanes.staleAfter { - return false - } - b.dropInbound(item, laneDropReasonStale, "waited past the staleness threshold") - return true -} - -// laneStaleEligible 报告某条入站是否受陈旧度闸约束。两类**豁免**各有硬理由: -// -// - **非聊天事件**(notice / request / meta_event / 解析失败兜底,ContentType=action): -// 它们不产生「回答旧话」这种失真,却是渠道事件留存(design/26 §5)与审计的唯一来源。 -// 按年龄丢掉它们只会让留存出现空洞。 -// - **主人私聊**:「回主人私聊」不是对群聊话题的应答,迟到的回答仍然是回答; -// 丢它等于对主人静默,直接违反「该回的回」这条红线。 -func laneStaleEligible(item laneItem) bool { - if item.msg.ContentType == adapter.ContentAction { - return false - } - if !item.msg.IsGroup() && item.actor.IsOwner() { - return false - } - return true -} - -// dropInbound 给「被丢弃的入站」留下**三样痕迹**,缺一不可: -// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上 -// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 P2-3: -// 计数只活在进程内存里等于没有出口); -// - 按原因的累计计数(LaneStats,供进程内调用方/后续面板接线读,不新开端点); -// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。 -// -// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。 -func (b *Bridge) dropInbound(item laneItem, reason, detail string) { - stale, overflow := b.lanes.noteDrop(reason) - b.lanes.log.Warn("inbound dropped by session lane", - "reason", reason, "detail", detail, - "platform", item.msg.Platform, "channel", item.msg.ChannelID, - "sender", item.msg.SenderID, "message_id", item.msg.MessageID, - "waited", b.now().Sub(item.arrivedAt).String(), - "dropped_stale_total", stale, "dropped_overflow_total", overflow) - if b.onRejected != nil { - b.onRejected(item.msg, reason) - } -} - -// noteDrop 累加某一类丢弃的计数,并返回累加后的两类计数快照(写进 WARN 用)。 -func (s *laneSet) noteDrop(reason string) (stale, overflow int64) { - s.mu.Lock() - defer s.mu.Unlock() - switch reason { - case laneDropReasonStale: - s.droppedStale++ - case laneDropReasonOverflow: - s.droppedOverflow++ - } - return s.droppedStale, s.droppedOverflow -} - -// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。 -// -// 它必须与 `offer` / `requeueFront` 用同一把锁:与 offer 交错的话会出现「offer 拿到 -// 车道 → worker 退休并删除 → offer 往一条没有 worker 的车道入队」的丢失。入队与 -// 「查空」在同一把锁下,就不存在这个窗口。 -func (b *Bridge) retireLaneIfIdle(l *lane) bool { - b.lanes.mu.Lock() - defer b.lanes.mu.Unlock() - if len(l.front) > 0 || len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l { - return false - } - delete(b.lanes.lanes, l.key) - return true -} - -// abandonLaneAtStop 登记「桥停止时车道里还排着、因此永远不会被回答的条目」。 -// -// 它与 coalesce_lane.go 的 abandonAtStop 分工明确:那边管**已经取在手上、正在等窗口** -// 的那一批,这里管**还躺在队列里**的(front/ch/spill)。两处都带 source 落 WARN 并计入 -// abandoned_at_stop_total —— 进程要停了没法「晚回」,唯一能做对的是**说出来**。 -func (b *Bridge) abandonLaneAtStop(l *lane) { - b.lanes.mu.Lock() - remaining := len(l.front) + len(l.ch) + len(l.spill) - b.lanes.mu.Unlock() - if remaining == 0 { - return - } - total := b.coalesceStats.abandonedAtStop.Add(int64(remaining)) - b.lanes.log.Warn("lane queue abandoned at bridge stop", - "lane", l.key, "count", remaining, "abandoned_at_stop_total", total) -} - -// logLaneStart 记「这一批等了多久才开始跑」—— 「她为什么回得晚」在生产日志里最直接 -// 的那个答案。与它同时出现在日志里的 `msg="inbound frame"`(适配器的入站帧 DEBUG) -// 则说明读循环在等的同时一直在读:两行合起来就是「回合没有堵住心跳」的证据。 -func (b *Bridge) logLaneStart(item laneItem, waited time.Duration, messages int) { - b.lanes.log.Debug("lane turn started", - "waited", waited.String(), "messages", messages, - "platform", item.msg.Platform, "channel", item.msg.ChannelID, - "sender", item.msg.SenderID, "group", item.msg.GroupID) -} - -// laneStrangerPrivate 报告一条入站是否属于**陌生人私聊**(非群 + 非主人)。 -// -// 它与 `isStrangerPrivate(sess)` 同源(同一判据的两面):会话层从 actor 派生 -// `sess.Owner`,这里直接用 actor 元数据,因此不需要先查会话就能判定「这条车道 -// 要不要受 applyToStrangers 管」。主人私聊走不到这里(那是 direct 车道)。 -func laneStrangerPrivate(item laneItem) bool { - return !item.msg.IsGroup() && !item.actor.IsOwner() -} - -// retireLane 无条件把车道从注册表摘掉(桥停止时的清理;未处理的条目随取消作废)。 -func (b *Bridge) retireLane(l *lane) { - b.lanes.mu.Lock() - defer b.lanes.mu.Unlock() - if b.lanes.lanes[l.key] == l { - delete(b.lanes.lanes, l.key) - } -} - -// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与 -// **会话键同源**(M22 复审 P1-1):两套各算各的判据会让同一个 session 出现两条车道、 -// 两个回合并发,而两条 run 共享 sessionID ⇒ checkpoint 行 id `{sessionID}:{step}` 互相 -// 覆盖、`loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用,flow 桶 -// `p.graphs[sessionID]` 被先收尾者删掉(loop 没有会话级互斥,IM 侧也没有网关那层 -// `session_turn_fence`)。 -// -// - **主人私聊**:无论来自哪个平台,都被 inboundTargetFor 路由进同一个 Web 主会话 -// (history.MainSessionID),因此共用一条 `main` 车道。这里的键与会话键**故意不同** -// ——`main` 是那条路由的目的地,比渠道键更强(把各平台的主人私聊并成一条)。 -// - 其余:**逐字就是** `session.Key(msg)`(`platform \x00 channelID`,与会话层缓存键 -// 同一个出处),所以「会话层认为是一个会话」与「车道认为是一个会话」不可能分叉。 -// -// 🔴 **取舍(比旧实现少了一个维度,是有意的)**:键里**没有账号名**。同一平台的两个账号 -// 收到同一对端/同一个群的消息时,它们是**同一个 session**(会话键只含 platform 与 -// channel),因此必须共用一条车道 —— 分开就会让同一会话两个回合并发。副作用是同一对端 -// 的多个账号的入站会串行排队;回复出口不受影响:出站路由取的是**每条消息自己的** -// `msg.Platform`/`ChannelID`/`GroupID` 与它自己的适配器(`laneItem.a`),车道本身从不 -// 持有任何账号级状态。 -// -// 同理也**不要**在这里重新引入「频道」维度:kook 同一 guild 的两个 text channel 在会话层 -// 是同一个群会话(键按 `group:`),按 `ChannelID` 分车道同样会造成双车道。 -func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor) string { - if ownerPrivateLane(msg, actor) { - return ownerPrivateLaneKey - } - if session.ChannelKey(msg) == "" { - // 没有任何渠道可用(私聊既无 ChannelID 也无 SenderID):会话层会以 - // ErrInvalidChannel 拒掉这条消息,这里只需要一个不会与真实会话撞键的稳定键。 - return "unbound" + laneKeySep + adapter.ConversationKey(msg) - } - return session.Key(msg) -} diff --git a/internal/services/channelbridge/lane_drop.go b/internal/services/channelbridge/lane_drop.go new file mode 100644 index 0000000..7df4f52 --- /dev/null +++ b/internal/services/channelbridge/lane_drop.go @@ -0,0 +1,131 @@ +package channelbridge + +// 本文件承载车道的**丢弃与停机诚实**:陈旧度闸、门前拒收、以及「桥停了,这些东西一条 +// 都没跑」的登记。它是「丢弃是诚实且有界」的落点 —— 每一类丢弃都留 +// WARN + 计数 + onRejected 审计,绝不静默消失(与 flowgate 的 rate_limited / +// review_pruned 同一条流水)。回什么属于回合语义,本文件一个字都不改。 + +import ( + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// LaneStats 是车道的计数快照:**被丢弃的入站有多少、为什么**(不变量 ③ 的可审计面), +// 加上主人私聊溢出区的当前积压。 +type LaneStats struct { + // ActiveLanes 是此刻有排队或在跑的会话车道数。 + ActiveLanes int + // DroppedStale 是因陈旧度闸被丢弃的条目累计数(超龄不补答)。 + DroppedStale int64 + // DroppedOverflow 是车道满时**在门前被拒收**的条目累计数。它只可能落在群聊 / + // 陌生人私聊上:主人私聊车道永不拒收(见 offer 的 neverDrop)。 + DroppedOverflow int64 + // Backlog 是主人私聊溢出区里此刻还排着的条目数(其余车道恒为 0)。 + Backlog int +} + +// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。 +// +// 生产可见性的口径(复审 P2-3):**不新开 HTTP 端点**(那是契约变更,归塔)。真正 +// 生产可见的是每一条丢弃 WARN 里带的 `dropped_stale_total` / `dropped_overflow_total` +// (累计计数随日志落地)以及 `onRejected` 审计轨;本方法只是给进程内调用方(管理面板 +// 将来接线、运维探针、测试)读同一组数字的窄口,不构成对外契约。 +func (b *Bridge) LaneStats() LaneStats { + if b.lanes == nil { + return LaneStats{} + } + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + backlog := 0 + for _, l := range b.lanes.lanes { + backlog += len(l.spill) + } + return LaneStats{ + ActiveLanes: len(b.lanes.lanes), + DroppedStale: b.lanes.droppedStale, + DroppedOverflow: b.lanes.droppedOverflow, + Backlog: backlog, + } +} + +// staleInbound 实现陈旧度闸:轮到时已超龄 ⇒ 丢弃不补答,并留 WARN + 计数 + 审计。 +// 返回 true 表示这条已被丢弃。waited 由调用方算好(含等全局回合槽的时间)。 +func (b *Bridge) staleInbound(item laneItem, waited time.Duration) bool { + if !laneStaleEligible(item) { + return false + } + if waited <= b.lanes.staleAfter { + return false + } + b.dropInbound(item, laneDropReasonStale, "waited past the staleness threshold") + return true +} + +// laneStaleEligible 报告某条入站是否受陈旧度闸约束。两类**豁免**各有硬理由: +// +// - **非聊天事件**(notice / request / meta_event / 解析失败兜底,ContentType=action): +// 它们不产生「回答旧话」这种失真,却是渠道事件留存(design/26 §5)与审计的唯一来源。 +// 按年龄丢掉它们只会让留存出现空洞。 +// - **主人私聊**:「回主人私聊」不是对群聊话题的应答,迟到的回答仍然是回答; +// 丢它等于对主人静默,直接违反「该回的回」这条红线。 +func laneStaleEligible(item laneItem) bool { + if item.msg.ContentType == adapter.ContentAction { + return false + } + if !item.msg.IsGroup() && item.actor.IsOwner() { + return false + } + return true +} + +// dropInbound 给「被丢弃的入站」留下**三样痕迹**,缺一不可: +// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上 +// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 P2-3: +// 计数只活在进程内存里等于没有出口); +// - 按原因的累计计数(LaneStats,供进程内调用方/后续面板接线读,不新开端点); +// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。 +// +// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。 +func (b *Bridge) dropInbound(item laneItem, reason, detail string) { + stale, overflow := b.lanes.noteDrop(reason) + b.lanes.log.Warn("inbound dropped by session lane", + "reason", reason, "detail", detail, + "platform", item.msg.Platform, "channel", item.msg.ChannelID, + "sender", item.msg.SenderID, "message_id", item.msg.MessageID, + "waited", b.now().Sub(item.arrivedAt).String(), + "dropped_stale_total", stale, "dropped_overflow_total", overflow) + if b.onRejected != nil { + b.onRejected(item.msg, reason) + } +} + +// noteDrop 累加某一类丢弃的计数,并返回累加后的两类计数快照(写进 WARN 用)。 +func (s *laneSet) noteDrop(reason string) (stale, overflow int64) { + s.mu.Lock() + defer s.mu.Unlock() + switch reason { + case laneDropReasonStale: + s.droppedStale++ + case laneDropReasonOverflow: + s.droppedOverflow++ + } + return s.droppedStale, s.droppedOverflow +} + +// abandonLaneAtStop 登记「桥停止时车道里还排着、因此永远不会被回答的条目」。 +// +// 它与 coalesce_lane.go 的 abandonAtStop 分工明确:那边管**已经取在手上、正在等窗口** +// 的那一批,这里管**还躺在队列里**的(front/ch/spill)。两处都带 source 落 WARN 并计入 +// abandoned_at_stop_total —— 进程要停了没法「晚回」,唯一能做对的是**说出来**。 +func (b *Bridge) abandonLaneAtStop(l *lane) { + b.lanes.mu.Lock() + remaining := len(l.front) + len(l.ch) + len(l.spill) + b.lanes.mu.Unlock() + if remaining == 0 { + return + } + total := b.coalesceStats.abandonedAtStop.Add(int64(remaining)) + b.lanes.log.Warn("lane queue abandoned at bridge stop", + "lane", l.key, "count", remaining, "abandoned_at_stop_total", total) +} diff --git a/internal/services/channelbridge/lane_owner.go b/internal/services/channelbridge/lane_owner.go new file mode 100644 index 0000000..189473a --- /dev/null +++ b/internal/services/channelbridge/lane_owner.go @@ -0,0 +1,68 @@ +package channelbridge + +// 本文件承载「**主人私聊**」这条车道的判据与车道键派生(M22 切片 LANEQUEUE; +// M23 的红线「主人私聊不要应用静默窗口」建立在这里的判据上)。 +// +// 判据只用 actor 元数据 + 白名单:**绝不读正文**、绝不看提示词(项目红线)。 +// +// 注意判据是 M22 既有的 `!msg.IsGroup() && actor.IsOwner()`,**不是**任务书里写的 +// `actor.IsGroupChat()`:两者在「适配器把群事件标成 Sender.IsGroup 但 ChannelType 仍是 +// Direct」这类帧上可能分叉(那条帧会走 direct,更保守,红线方向不受影响)。 +// 本片不改这个判据(改它会动 M22 的车道键与丢弃策略,属另一片)。 +// 它是 `neverDrop`(永不拒收)与 `direct`(无窗口、拿不到定时器工厂)两件事的 +// **同一个**来源,因此不存在「键说是主人、丢弃策略说不是」的两套口径。 + +import ( + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" +) + +// ownerPrivateLane 报告这条消息是否属于「主人私聊」车道 —— 也就是**永不丢弃**的那条。 +// 判据与 laneKey 完全同源(私聊 + actor 元数据说他是主人),因此不会出现「键说是主人、 +// 丢弃策略说不是」的两套口径。这里绝不读正文:身份只来自 actor 元数据 + 白名单。 +func ownerPrivateLane(msg adapter.UnifiedMessage, actor adapter.Actor) bool { + return !msg.IsGroup() && actor.IsOwner() +} + +// laneStrangerPrivate 报告一条入站是否属于**陌生人私聊**(非群 + 非主人)。 +// +// 它与 `isStrangerPrivate(sess)` 同源(同一判据的两面):会话层从 actor 派生 +// `sess.Owner`,这里直接用 actor 元数据,因此不需要先查会话就能判定「这条车道 +// 要不要受 applyToStrangers 管」。主人私聊走不到这里(那是 direct 车道)。 +func laneStrangerPrivate(item laneItem) bool { + return !item.msg.IsGroup() && !item.actor.IsOwner() +} + +// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与 +// **会话键同源**(M22 复审 P1-1):两套各算各的判据会让同一个 session 出现两条车道、 +// 两个回合并发,而两条 run 共享 sessionID ⇒ checkpoint 行 id `{sessionID}:{step}` 互相 +// 覆盖、`loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用,flow 桶 +// `p.graphs[sessionID]` 被先收尾者删掉(loop 没有会话级互斥,IM 侧也没有网关那层 +// `session_turn_fence`)。 +// +// - **主人私聊**:无论来自哪个平台,都被 inboundTargetFor 路由进同一个 Web 主会话 +// (history.MainSessionID),因此共用一条 `main` 车道。这里的键与会话键**故意不同** +// ——`main` 是那条路由的目的地,比渠道键更强(把各平台的主人私聊并成一条)。 +// - 其余:**逐字就是** `session.Key(msg)`(`platform \x00 channelID`,与会话层缓存键 +// 同一个出处),所以「会话层认为是一个会话」与「车道认为是一个会话」不可能分叉。 +// +// 🔴 **取舍(比旧实现少了一个维度,是有意的)**:键里**没有账号名**。同一平台的两个账号 +// 收到同一对端/同一个群的消息时,它们是**同一个 session**(会话键只含 platform 与 +// channel),因此必须共用一条车道 —— 分开就会让同一会话两个回合并发。副作用是同一对端 +// 的多个账号的入站会串行排队;回复出口不受影响:出站路由取的是**每条消息自己的** +// `msg.Platform`/`ChannelID`/`GroupID` 与它自己的适配器(`laneItem.a`),车道本身从不 +// 持有任何账号级状态。 +// +// 同理也**不要**在这里重新引入「频道」维度:kook 同一 guild 的两个 text channel 在会话层 +// 是同一个群会话(键按 `group:`),按 `ChannelID` 分车道同样会造成双车道。 +func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor) string { + if ownerPrivateLane(msg, actor) { + return ownerPrivateLaneKey + } + if session.ChannelKey(msg) == "" { + // 没有任何渠道可用(私聊既无 ChannelID 也无 SenderID):会话层会以 + // ErrInvalidChannel 拒掉这条消息,这里只需要一个不会与真实会话撞键的稳定键。 + return "unbound" + laneKeySep + adapter.ConversationKey(msg) + } + return session.Key(msg) +} diff --git a/internal/services/channelbridge/lane_queue.go b/internal/services/channelbridge/lane_queue.go new file mode 100644 index 0000000..93543ba --- /dev/null +++ b/internal/services/channelbridge/lane_queue.go @@ -0,0 +1,246 @@ +package channelbridge + +// 本文件承载车道的**入队与取队列**(M22 切片 LANEQUEUE;M23 加了 batch 车道的容量与 +// 回插队首)。它是「读与回水解耦」的那一侧:读循环只投递,worker 只按 FIFO 取。 +// +// 三条不变量落在这里: +// - **有界且诚实**:群聊/陌生人满时在门前拒收**这一条**(队列里的条目一条都不动), +// 主人私聊永不拒收(溢到 spill 只排队,按倍率告警); +// - **FIFO 不可超车**:spill 一旦非空就一直追加 spill;front(回插区)里的一定早于 +// ch 与 spill 里的,因此读取顺序恒为 front → ch → spill; +// - **整片取出**:`takeAll` 是静默窗口攒批的唯一取队列入口(只搬不筛、不改序)。 + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// enqueueInbound 把一条**已过流控闸**的入站消息投进它所属会话的车道,并立刻返回 —— +// 这是「读与回水解耦」的全部:适配器的读循环调用它之后马上可以继续读 socket。 +// +// 🔴 这就是 M23 静默窗口合并的**入队 seam**:入参是一条**已过闸门的 prepared item** +// (Validate/ResolveActor/Admit 都已在本函数的调用方 onMessage 里做完,连同解析结果 +// actor 一起带进来)。车道头部只做「等(窗口)+ 发车」,准入判断不在这里。 +// +// 三件事在这里一次性做完: +// - 认领车道(没有就建一条并起 worker)—— **类型由 actor 元数据决定**: +// 主人私聊 ⇒ direct(无窗口),其余 ⇒ batch(有窗口); +// - 入队(群聊/陌生人满则**在门前拒收这一条**并留痕;主人私聊**永不拒收**); +// - 与 Stop 的 `wg.Wait` 对齐:`b.wg.Add` 与 `closed` 的判定在同一把 `b.mu` 下, +// 因此 Stop 要么看见这个工作项、要么我们看见 closed 而不入队,二者必居其一。 +// +// 锁顺序:全树只有这一处同时取两把锁,顺序恒为 `b.mu` → `lanes.mu`。 +func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, actor adapter.Actor) { + if b.lanes == nil { + return + } + item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()} + key := laneKey(msg, actor) + // 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**: + // 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。 + neverDrop := ownerPrivateLane(msg, actor) + spec := b.laneSpecFor(key, item, neverDrop) + + b.mu.Lock() + if b.closed { + b.mu.Unlock() + b.lanes.log.Debug("bridge closed, inbound not queued", + "platform", msg.Platform, "channel", msg.ChannelID) + return + } + l, start, rejected := b.lanes.offer(key, item, spec) + if start { + b.wg.Add(1) + } + b.mu.Unlock() + + if start { + go b.laneWorker(ctx, l) + } + if rejected { + b.dropInbound(item, laneDropReasonOverflow, "lane queue full") + } +} + +// laneSpecFor 决定一条车道的形状:**类型只由 actor 元数据决定** +// (`IsOwner() && !IsGroupChat()` ⇒ direct),窗口参数只影响容量。 +// +// 容量:batch 车道至少要能装下一整批,否则「一批 N 条」在结构上不可达 +// (队列深度 8 时第 9 条就会在门前被拒收,合并就成了摆设)。因此 +// capacity = max(默认深度, 生效的单批上限);direct 车道用默认深度。 +// 内存口径因此是「活跃会话数 × 批上限」,而批上限本身被 clamp 在 [1,50]。 +func (b *Bridge) laneSpecFor(key string, item laneItem, neverDrop bool) laneSpec { + spec := laneSpec{neverDrop: neverDrop, batch: !neverDrop, capacity: b.lanes.depth} + if neverDrop { + return spec + } + if w, ok := b.coalesceWindowFor(key, laneStrangerPrivate(item)); ok && w.maxBatch > spec.capacity { + spec.capacity = w.maxBatch + } + return spec +} + +// offer 在**一个临界区**内完成「查/建车道 + 入队」,使 `retireLaneIfIdle` 不可能夹在 +// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。 +// +// 队列满时的策略分两档(都不是静默的): +// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按倍率 +// (ownerLaneBacklogWarn 及其翻倍)留 WARN —— 一个白名单账号的积压只反映「她落后了」, +// 不构成第三方洪泛面; +// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里 +// 的条目因此一条都不会凭空消失 —— 窗口的 takeAll 依赖这一点;被拒收的那条 +// 由调用方留痕(WARN + 计数 + 审计)。 +// +// 🔴 **FIFO 的关键**:`spill` 一旦非空,后面的条目**一律继续追加到 spill**,不再回写 +// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车(复审实测顺序 +// `[m1 m2 m3 m5 m4]`)。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」: +// spill 里的一定晚于 ch 里的,而 spill 一旦开始就用到底(直到它被取空)。 +// +// 同理,`front`(回插队首区)里的一定**早于** ch 与 spill 里的(只有 worker 自己 +// 在发车前把超量的尾部放进去),所以读取顺序恒为 front → ch → spill。 +func (s *laneSet) offer(key string, item laneItem, spec laneSpec) (l *lane, start bool, rejected bool) { + s.mu.Lock() + defer s.mu.Unlock() + l = s.lanes[key] + if l == nil { + capacity := spec.capacity + if capacity <= 0 { + capacity = s.depth + } + ch := make(chan laneItem, capacity) + if spec.batch { + l = newBatchLane(key, ch, spec.neverDrop, &batchWindow{after: s.after}) + } else { + l = newDirectLane(key, ch, spec.neverDrop) + } + s.lanes[key] = l + start = true + } + if l.neverDrop && len(l.spill) > 0 { + s.spillLocked(l, item) + return l, start, false + } + select { + case l.ch <- item: + return l, start, false + default: + } + if l.neverDrop { + s.spillLocked(l, item) + return l, start, false + } + return l, start, true +} + +// spillLocked 把条目追加进主人私聊车道的溢出区,并按**倍率**告警 +// (16/32/64/128… 每翻一倍一条 WARN):固定水位只会在恰好等于时响一次,之后继续涨就 +// 再无信号,而「她落后了多少」正是主人侧唯一可观测的量。调用方必须持有 s.mu。 +func (s *laneSet) spillLocked(l *lane, item laneItem) { + l.spill = append(l.spill, item) + if n := len(l.spill); n >= ownerLaneBacklogWarn && isPowerOfTwo(n) { + s.log.Warn("owner private lane backlog is growing, still queueing", + "backlog", n, "lane", l.key, + "platform", item.msg.Platform, "channel", item.msg.ChannelID) + } +} + +// isPowerOfTwo 报告 n 是否为 2 的幂(n ≥ 1)。 +func isPowerOfTwo(n int) bool { return n > 0 && n&(n-1) == 0 } + +// nextLaneItem 按 FIFO 取下一条:先取回插队首区(front),再把缓冲区取完,最后取 +// 主人私聊的溢出区(三者的先后由 offer / requeueFront 的写入口径保证)。 +// ok=false 表示此刻真的没有待办。 +func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) { + b.lanes.mu.Lock() + if len(l.front) > 0 { + item := l.front[0] + l.front = l.front[1:] + b.lanes.mu.Unlock() + return item, true + } + b.lanes.mu.Unlock() + select { + case item := <-l.ch: + return item, true + default: + } + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if len(l.spill) == 0 { + return laneItem{}, false + } + item := l.spill[0] + copy(l.spill, l.spill[1:]) + l.spill = l.spill[:len(l.spill)-1] + return item, true +} + +// takeAll 把车道上**此刻**排着的条目整片取出(顺序 = 到达顺序),供静默窗口的 +// 批组装使用。它是合并阶段**唯一**的取队列入口(M23 蓝图 §4-1): +// +// - 只有车道的单消费者(该车道唯一的 worker)调用它 ⇒ 不存在两个消费者各抓一半; +// - 取出即「这批就是这些」,等待期间新到的继续落进队列,下一轮 takeAll 再取; +// - 它**不**做任何筛选:不丢、不重排、不改内容(窗口内零丢失的结构保证)。 +func (b *Bridge) takeAll(l *lane) []laneItem { + b.lanes.mu.Lock() + out := l.front + l.front = nil + b.lanes.mu.Unlock() + for { + select { + case item := <-l.ch: + out = append(out, item) + continue + default: + } + b.lanes.mu.Lock() + if len(l.spill) > 0 { + out = append(out, l.spill...) + l.spill = nil + } + b.lanes.mu.Unlock() + return out + } +} + +// requeueFront 把一批**尚未发车**的条目按原顺序插回队首,成为下一批。 +// +// 它是「单批上限不是丢弃」的落点(M23 蓝图 §4-3):超量的**尾部**原样回插、保序、 +// 在锁内、在发车之前。回插到 front 而不是写 ch:ch 有界且写满即拒收, +// 而 front 无界但只由 worker 在发车前写入,写入量与「刚拆出来的尾部」相等。 +func (b *Bridge) requeueFront(l *lane, items []laneItem) { + if len(items) == 0 { + return + } + b.lanes.mu.Lock() + merged := make([]laneItem, 0, len(items)+len(l.front)) + merged = append(merged, items...) + merged = append(merged, l.front...) + l.front = merged + b.lanes.mu.Unlock() +} + +// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。 +// +// 它必须与 `offer` / `requeueFront` 用同一把锁:与 offer 交错的话会出现「offer 拿到 +// 车道 → worker 退休并删除 → offer 往一条没有 worker 的车道入队」的丢失。入队与 +// 「查空」在同一把锁下,就不存在这个窗口。 +func (b *Bridge) retireLaneIfIdle(l *lane) bool { + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if len(l.front) > 0 || len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l { + return false + } + delete(b.lanes.lanes, l.key) + return true +} + +// retireLane 无条件把车道从注册表摘掉(桥停止时的清理;未处理的条目随取消作废)。 +func (b *Bridge) retireLane(l *lane) { + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if b.lanes.lanes[l.key] == l { + delete(b.lanes.lanes, l.key) + } +} diff --git a/internal/services/channelbridge/lane_worker.go b/internal/services/channelbridge/lane_worker.go new file mode 100644 index 0000000..3f6ad1e --- /dev/null +++ b/internal/services/channelbridge/lane_worker.go @@ -0,0 +1,108 @@ +package channelbridge + +// 本文件承载车道的**单消费者**:一个会话一个 worker,串行跑完队列里的每一条。 +// +// 「回合在跑」在结构上就等于「worker 正忙」—— M23 的回合中续跑因此不需要任何 +// turn-in-flight 标志:回合结束回到循环头,takeAll 自然把这段期间攒下的消息整片取出。 +// +// direct 车道(主人私聊)在这里按 `l.batcher == nil` 分派到 runDirectLaneItem: +// 到达即投递、一条一条跑,**没有**任何窗口等待。 + +import ( + "context" + "time" +) + +// laneWorker 是某会话车道的唯一执行者:串行跑完每一条,队列空了就把车道交还注册表 +// (空闲车道不占内存,也不会把「这个人再说话」变慢)。 +// +// **回合中续跑(⑤)就落在这里**:不存在任何 "turn in flight" 标志 —— 「回合在跑」 +// 在结构上就等于「worker 正忙」,跑回合期间新消息落进同一个队列,回合结束回到循环头, +// takeAll 自然把这段期间攒下的消息整片取出 ⇒ **一轮续跑**(不是 N 轮,也不是 0 条)。 +func (b *Bridge) laneWorker(ctx context.Context, l *lane) { + defer b.wg.Done() + for { + // 桥已停:队列里还没发车的条目**如实登记**(绝不静默消失),然后退休。 + if ctx.Err() != nil { + b.abandonLaneAtStop(l) + b.retireLane(l) + return + } + if item, ok := b.nextLaneItem(l); ok { + b.dispatchLaneItem(ctx, l, item) + continue + } + // 队列空了才谈退休:摘表与入队用同一把锁,因此「摘下」与「刚有人塞进来」不会 + // 交错成条目丢失(见 retireLaneIfIdle 的说明)。 + if b.retireLaneIfIdle(l) { + return + } + select { + case <-ctx.Done(): + // 桥停了:把车道摘掉,未处理的条目随进程取消一起作废(Stop 的语义)。 + b.retireLane(l) + return + case item := <-l.ch: + b.dispatchLaneItem(ctx, l, item) + } + } +} + +// dispatchLaneItem 按**车道类型**分派:direct(主人私聊)一条一条跑;batch 走窗口 +// 引擎(见 coalesce_lane.go 的 runBatchLaneItem)。 +// +// 分派依据是 `l.batcher == nil` —— 一个**结构**判据,不是配置开关:direct 车道的 +// 类型里根本没有 batcher 字段值,窗口代码对它不可达。 +func (b *Bridge) dispatchLaneItem(ctx context.Context, l *lane, item laneItem) { + if l.batcher == nil { + b.runDirectLaneItem(ctx, l, item) + return + } + b.runBatchLaneItem(ctx, l, item) +} + +// runDirectLaneItem 跑 direct 车道(= 主人私聊)的一条:先取全局回合槽(**在线程池里 +// 等,不在读循环里等**),再过陈旧度闸,然后走逐条闸门 + 单条回合。**没有任何等待窗口**。 +func (b *Bridge) runDirectLaneItem(ctx context.Context, l *lane, item laneItem) { + if !b.acquireTurn(ctx) { + return + } + defer b.releaseTurn() + + waited := b.now().Sub(item.arrivedAt) + if b.staleInbound(item, waited) { + return + } + b.logLaneStart(item, waited, 1) + p, ok := b.prepareInbound(ctx, item) + if !ok { + l.lastTurnEnd = b.now() + return + } + b.runBatch(ctx, []preparedInbound{p}) + l.lastTurnEnd = b.now() +} + +// acquireTurn 取一个全局回合槽(maxConcurrent 个)。ctx 取消时返回 false:桥停了 +// 就不再开新回合(carried 条目由 Stop 的诚实登记负责,见 coalesce_lane.go)。 +func (b *Bridge) acquireTurn(ctx context.Context) bool { + select { + case b.sem <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +// releaseTurn 归还一个全局回合槽。 +func (b *Bridge) releaseTurn() { <-b.sem } + +// logLaneStart 记「这一批等了多久才开始跑」—— 「她为什么回得晚」在生产日志里最直接 +// 的那个答案。与它同时出现在日志里的 `msg="inbound frame"`(适配器的入站帧 DEBUG) +// 则说明读循环在等的同时一直在读:两行合起来就是「回合没有堵住心跳」的证据。 +func (b *Bridge) logLaneStart(item laneItem, waited time.Duration, messages int) { + b.lanes.log.Debug("lane turn started", + "waited", waited.String(), "messages", messages, + "platform", item.msg.Platform, "channel", item.msg.ChannelID, + "sender", item.msg.SenderID, "group", item.msg.GroupID) +} diff --git a/internal/services/gateway/admin_coalesce.go b/internal/services/gateway/admin_coalesce.go index 953a957..dcea226 100644 --- a/internal/services/gateway/admin_coalesce.go +++ b/internal/services/gateway/admin_coalesce.go @@ -19,8 +19,9 @@ package gateway // 因此不是 owner-only;也不给 viewer(它是渠道调优,不是监控投影)。 // 3. **热生效**:桥每次发车前从**同一个** store 实例读生效值,因此保存后下一次发车 // 就用新值 —— 响应里的 `hot:true` 是事实,不是承诺。 -// 4. **主人私聊是只读事实**:车道类型由 actor 元数据决定(`IsOwner() && !IsGroupChat()` -// ⇒ direct 车道、**没有窗口对象、拿不到定时器工厂**),本端点**没有任何键**能翻转它。 +// 4. **主人私聊是只读事实**:车道类型由 actor 元数据决定(判据 `ownerPrivateLane`: +// `!msg.IsGroup() && actor.IsOwner()` ⇒ direct 车道、**没有窗口对象、拿不到定时器 +// 工厂**),本端点**没有任何键**能翻转它。 // `ownerPrivate:"never"` 就是这条事实的下发形状,面板据此只显示一行说明、不给开关。 // // 零新 `.env` 键(design/13 §2.1):配置落在 `$DATA_DIR/config/coalesce.json`。