Cyrene_v2/internal/services/channelbridge/lane_drop.go

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)
}