Cyrene_v2/internal/services/channelbridge/coalesce_lane.go

package channelbridge

// coalesce_lane.go 把静默窗口接到车道上（M23 的「车道头部」）：攒批 → 逐条陈旧度
// → 逐条闸门 → 单批上限拆分 → 一个 turnID 发车。它是 coalesce.go（窗口引擎）与
// coalesce_batch.go（一回合一 Reply）之间的那层「调度」。
//
// 本文件里**没有**任何「跳过某条消息」的分支，唯一的剔除来自 M22 的**既有**闸门：
//   - 陈旧度闸（staleInbound：轮到时已超龄 ⇒ 不补答，逐条 WARN + 计数 + 审计）；
//   - 逐条闸门（prepareInbound：黑白名单 / 静默 / 空内容 / 仅登记的通知，全是
//     「这条本来就不该驱动回合」，不是「批装不下所以扔掉」）。
//
// 装不下的**不是丢弃**：超过单批上限的尾部原样回插队首，成为下一批（requeueFront），
// 并且 CoalesceStats().CoalescedDroppedTotal 恒为 0 —— 它存在的意义就是让「有人不小心
// 加了一条 continue」有一个能被断言的出口。

import (
	"context"
)

// runBatchLaneItem 是 batch 车道的发车入口：把刚取到的 first 与此刻队列里的其余条目
// 合成一批，按窗口等待，然后交给 flushLaneBatch。
//
// 🔴 **回合中续跑（⑤）不需要额外标志**：worker 在跑回合时根本不在这里，新消息落进
// 同一个队列；回合结束回到 laneWorker 循环头，下一次 nextLaneItem + takeAll 就把
// 这段期间攒下的整片取出 ⇒ 一轮续跑（见 flushLaneBatch 的 continued 计数）。
func (b *Bridge) runBatchLaneItem(ctx context.Context, l *lane, first laneItem) {
	w, windowed := b.coalesceWindowFor(l.key, laneStrangerPrivate(first))
	if !windowed || !laneWindowAdmits(first.msg) {
		// 三种「到达即发车」：关窗（逐字回到 M22）、该会话不适用窗口、以及**不进窗口的帧**
		// （系统通知 / action 帧 / 我们自己的出站回执）。最后一种刻意**不**建窗口：
		// 一条回执不该让这一批的静默计时重新开始（M23 任务书的窗口准入条件）。
		// 刻意**不** takeAll：多取的条目会让「不合并」变成「悄悄合并」。
		b.flushLaneBatch(ctx, l, []laneItem{first}, w, false, coalesceFlushOff)
		return
	}
	items, solos := splitWindowItems(b.takeAll(l))
	items = append([]laneItem{first}, items...)
	var flushReason string
	var moreSolos []laneItem
	items, moreSolos, flushReason = b.coalesceWait(ctx, l, items, w)
	solos = append(solos, moreSolos...)
	if flushReason == coalesceFlushStop {
		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)
}

// flushLaneBatch 把一批**已经攒好**的条目发车（或按上限拆开）。
//
// 顺序是刻意的：**拆分**（结构，不影响任何一条的去向）→ **取全局回合槽** →
// 逐条陈旧度 → 逐条闸门 → 组上下文 → 一次 Run。拆分放在最前面，因为回插的尾部
// 还没到「发车那一刻」，它的陈旧度要在它自己那一批发车时判。
func (b *Bridge) flushLaneBatch(ctx context.Context, l *lane, items []laneItem, w coalesceWindow, windowed bool, reason string) {
	if len(items) == 0 {
		return
	}
	items = b.splitBatch(l, items, w, windowed)
	if !b.acquireTurn(ctx) {
		// 桥停了：这一批一条都没跑，必须如实登记（绝不静默消失）。
		b.abandonAtStop(l, items)
		return
	}
	defer b.releaseTurn()

	kept, stale := b.gateBatch(ctx, items)
	if len(kept) == 0 {
		if stale == len(items) {
			// 整批都超龄：不发车、留一条可查的痕迹（M22 的「太旧不补答」优先）。
			b.noteFlushReason(coalesceFlushAllStale)
			b.coalesceLog.Info("coalesce batch all stale, nothing to say",
				"lane", l.key, "count", len(items), "flush_reason", coalesceFlushAllStale)
			l.lastTurnEnd = b.now()
			return
		}
		// 一条都没进回合（全被逐条闸门静默/拒收）：这是「回什么」的既有裁决，
		// 不是合并丢的 —— 照样记一行，便于和上面的分支区分。
		b.coalesceLog.Debug("coalesce batch produced no turn",
			"lane", l.key, "count", len(items), "flush_reason", reason)
		l.lastTurnEnd = b.now()
		return
	}

	continued := !l.lastTurnEnd.IsZero() && items[0].arrivedAt.Before(l.lastTurnEnd)
	b.countDeparture(kept, reason, continued)
	b.logDeparture(l, items[0], kept, reason, continued)
	b.runBatchSegments(ctx, kept)
	l.lastTurnEnd = b.now()
}

// splitBatch 实施单批上限：超过上限时把**尾部**原样回插队首（保序、锁内、发车前），
// 本批只带走前 maxBatch 条。返回本批真正要发车的切片。
func (b *Bridge) splitBatch(l *lane, items []laneItem, w coalesceWindow, windowed bool) []laneItem {
	if !windowed || w.maxBatch <= 0 || len(items) <= w.maxBatch {
		return items
	}
	tail := items[w.maxBatch:]
	b.requeueFront(l, tail)
	b.coalesceStats.batchesSplit.Add(1)
	b.coalesceLog.Info("coalesce batch split, tail requeued (not dropped)",
		"lane", l.key, "kept", w.maxBatch, "requeued", len(tail), "max_batch", w.maxBatch)
	return items[:w.maxBatch]
}

// gateBatch 是**逐条闸门**的批量形式，不是合并期的筛选：它调用的两个判断（陈旧度闸 /
// 逐条闸门）在 M22 的单条流水线里本来就对每一条各跑一次，合并**没有新增任何丢弃点** ——
// 因此 CoalescedDroppedTotal 恒为 0（见 coalesce.go 的说明）。
//
// 它返回要进本轮的消息与「因超龄被丢」的条数。陈旧度必须在**组上下文之前**判：不能让
// 「批里最新一条是新鲜的」把超龄的旧条也塞进上下文（M22「太旧不补答」优先）；
// 被丢的每一条都由 dropInbound 留 WARN + 计数 + 审计（与单条路径逐字同一条流水）。
func (b *Bridge) gateBatch(ctx context.Context, items []laneItem) ([]preparedInbound, int) {
	kept := make([]preparedInbound, 0, len(items))
	stale := 0
	for _, item := range items {
		if b.staleInbound(item, b.now().Sub(item.arrivedAt)) {
			stale++
			continue
		}
		p, ok := b.prepareInbound(ctx, item)
		if !ok {
			continue
		}
		kept = append(kept, p)
	}
	return kept, stale
}

// countDeparture 记发车计数（批数/条数/发车原因/续跑）。计数与日志分开，
// 使「发车的形状」既能在日志里读、也能在测试里断言。
func (b *Bridge) countDeparture(kept []preparedInbound, reason string, continued bool) {
	b.coalesceStats.batchesTotal.Add(1)
	b.coalesceStats.batchMessagesTotal.Add(int64(len(kept)))
	if continued {
		b.coalesceStats.continuations.Add(1)
	}
	b.noteFlushReason(reason)
}

// logDeparture 记「这一批是怎么发车的」：条数、等了多久、发车原因、是否续跑。
// quiet/off 走 DEBUG，max_wait 走 INFO（它是「差点迟滞」的信号，值得在日常日志里看见）。
func (b *Bridge) logDeparture(l *lane, first laneItem, kept []preparedInbound, reason string, continued bool) {
	attrs := []any{
		"lane", l.key, "messages", len(kept),
		"waited_ms", b.now().Sub(first.arrivedAt).Milliseconds(),
		"flush_reason", reason, "continued", continued,
		"platform", first.msg.Platform, "channel", first.msg.ChannelID, "group", first.msg.GroupID,
	}
	if reason == coalesceFlushMaxWait {
		b.coalesceLog.Info("coalesce batch assembled (max wait reached)", attrs...)
		return
	}
	b.coalesceLog.Debug("coalesce batch assembled", attrs...)
}

// abandonAtStop 诚实登记「进程停机时这批一条都没发车」。主人 2026-10-24 的口径是
// 「宁可晚回，不许不回」，但进程已经要停了 —— 唯一能做对的事是**说出来**：
// 带 source 的 WARN + 累计计数（abandoned_at_stop_total），绝不静默消失。
func (b *Bridge) abandonAtStop(l *lane, items []laneItem) {
	if len(items) == 0 {
		return
	}
	total := b.coalesceStats.abandonedAtStop.Add(int64(len(items)))
	b.coalesceLog.Warn("coalesce batch abandoned at bridge stop",
		"lane", l.key, "count", len(items),
		"flush_reason", coalesceFlushStop, "abandoned_at_stop_total", total,
		"platform", items[0].msg.Platform, "channel", items[0].msg.ChannelID)
}