feat(channels): 陌生人私聊正常聊天（跳概率门）+ 强制流控 6条/分与回合槽位 + 安静时段不再硬静默 + 受限车道边界行前置
- 陌生人私聊：不再走「值不值得回」的概率门，唯一闸门 = 每发送者 6 条/分滑窗 + 同会话 1 个在跑回合 - 被限流只登记不回，且留带 source 的可见告警（不静默） - 安静时段在回复侧只作信号（规则层不再静默），群聊取消夜间静默 - 锁定流控常量：FlowDirectDefault 加字面量钉桩测试（放大它 = 静默关流控，且会让额度循环跑百万轮）
This commit is contained in:
AskaEth committed
1 parent
c6f5c3fea0
commit
7ae4298871
10 files changed
+654
-118
No files matched your search
@@ -44,6 +44,13 @@ type FlowGateOptions struct {
|
||||
// GroupMsgPerMin caps group-chat inbound messages per minute **per
|
||||
// conversation** (design/12 §6 策略三 / 裁决 06.4.3). Zero disables the limit.
|
||||
GroupMsgPerMin int
|
||||
// DirectMsgPerMin caps **direct (private)** inbound messages per minute
|
||||
// **per conversation** — and for a private message the conversation key is
|
||||
// (platform, channel, sender), so the limit throttles one sender rather than
|
||||
// the whole account. It is the mandatory flow control for stranger private
|
||||
// chats (主人 2026-10-05 裁决③): strangers chat normally, so a sender must be
|
||||
// unable to flood her. Zero disables the limit.
|
||||
DirectMsgPerMin int
|
||||
// MsgPerMin caps total inbound non-owner messages per minute across the whole
|
||||
// account — the anti-flood total guard. Zero disables it.
|
||||
MsgPerMin int
|
||||
@@ -76,6 +83,12 @@ const (
|
||||
// FlowLimitDefault is the protective default of 6 group messages per minute
|
||||
// (design/12 §6 策略三). It is applied when an account pins no group rate limit.
|
||||
FlowLimitDefault = 6
|
||||
// FlowDirectDefault is the protective default of 6 direct (private) messages
|
||||
// per minute **per sender** — the mandatory flow control for stranger private
|
||||
// chats (主人 2026-10-05 裁决③). 6/分钟与群聊保护性默认同一口径：一个人正常
|
||||
// 一来一回打字到不了 6 条/分钟，而刷屏/机器人会在第 7 条被挡下。它不随
|
||||
// groupMsgPerMin 变化（陌生人流控不允许被「没配那项」静默关掉）。
|
||||
FlowDirectDefault = 6
|
||||
// FlowCapDefault is the protective default total-cap of 50 pending messages
|
||||
// (design/12 §6 策略四).
|
||||
FlowCapDefault = 50
|
||||
@@ -113,16 +126,22 @@ func ConversationKey(msg UnifiedMessage) string {
|
||||
// 每个会话保存一份，因此两个群并发互不挤占（裁决 06.4.3）。
|
||||
type flowSession struct {
|
||||
groupHits []time.Time
|
||||
directHits []time.Time
|
||||
buffer []queuedMsg
|
||||
lastReview time.Time
|
||||
// inFlight 是该会话**正在跑**的回合数（AcquireTurn/ReleaseTurn 维护）。受限车道
|
||||
// 用它把「同一发送者同时最多 1 个在跑的回合」落到实处。
|
||||
inFlight int
|
||||
}
|
||||
|
||||
// FlowGate is a concurrency-safe, per-account ingress gate implementing the two
|
||||
// measures from design/12 §6 策略三/四:
|
||||
//
|
||||
// - 策略三: a sliding-window rate limit on group chat (and, optionally, all
|
||||
// non-owner messages). Over-limit messages are dropped (downgraded) rather
|
||||
// than blocking a flood of messages from saturating the loop.
|
||||
// non-owner messages), plus the mandatory **per-sender** rate limit on direct
|
||||
// (private) messages that backs the stranger private-chat flow control
|
||||
// (AllowDirect). Over-limit messages are dropped (downgraded) rather than
|
||||
// blocking a flood of messages from saturating the loop.
|
||||
// - 策略四: a bounded pending-message set. When it nears its cap the Reviewer is
|
||||
// invoked to prune low-value messages; with no Reviewer the oldest messages are
|
||||
// evicted FIFO. Either way every dropped message is audited.
|
||||
@@ -131,15 +150,18 @@ type flowSession struct {
|
||||
//
|
||||
// Owner-private messages are not routed through the gate (the caller bypasses it),
|
||||
// so the owner is never throttled or pruned. Identity is never examined here.
|
||||
// The gate also carries the per-conversation running-turn slot (AcquireTurn /
|
||||
// ReleaseTurn) used by the stranger private-chat flow control.
|
||||
type FlowGate struct {
|
||||
groupPerMin int
|
||||
msgPerMin int
|
||||
maxMessages int
|
||||
reviewAt int
|
||||
window time.Duration
|
||||
review Reviewer
|
||||
onPruned func(UnifiedMessage, string)
|
||||
now func() time.Time
|
||||
groupPerMin int
|
||||
directPerMin int
|
||||
msgPerMin int
|
||||
maxMessages int
|
||||
reviewAt int
|
||||
window time.Duration
|
||||
review Reviewer
|
||||
onPruned func(UnifiedMessage, string)
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
totalHits []time.Time
|
||||
@@ -160,14 +182,15 @@ type queuedMsg struct {
|
||||
// except Window which defaults to one minute when left zero.
|
||||
func NewFlowGate(opts FlowGateOptions) *FlowGate {
|
||||
g := &FlowGate{
|
||||
groupPerMin: opts.GroupMsgPerMin,
|
||||
msgPerMin: opts.MsgPerMin,
|
||||
maxMessages: opts.MaxMessages,
|
||||
window: opts.Window,
|
||||
review: opts.Review,
|
||||
onPruned: opts.OnPruned,
|
||||
now: opts.Now,
|
||||
sessions: make(map[string]*flowSession),
|
||||
groupPerMin: opts.GroupMsgPerMin,
|
||||
directPerMin: opts.DirectMsgPerMin,
|
||||
msgPerMin: opts.MsgPerMin,
|
||||
maxMessages: opts.MaxMessages,
|
||||
window: opts.Window,
|
||||
review: opts.Review,
|
||||
onPruned: opts.OnPruned,
|
||||
now: opts.Now,
|
||||
sessions: make(map[string]*flowSession),
|
||||
}
|
||||
if g.window <= 0 {
|
||||
g.window = defaultRateWindow
|
||||
@@ -258,6 +281,61 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
|
||||
return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}
|
||||
}
|
||||
|
||||
// AllowDirect enforces the per-sender sliding-window rate limit for a **direct
|
||||
// (private)** message (主人 2026-10-05 裁决③：陌生人私聊=正常聊天，因此必须有流控).
|
||||
// For a private message the conversation key is (platform, channel, sender), so
|
||||
// the window throttles one sender instead of the whole account and two strangers
|
||||
// never crowd each other out. It returns false when that sender is over
|
||||
// DirectMsgPerMin; the refused message is recorded to the OnPruned audit hook so
|
||||
// nothing is dropped silently. A zero limit or a group message always allows.
|
||||
func (g *FlowGate) AllowDirect(msg UnifiedMessage) bool {
|
||||
if g.directPerMin <= 0 || msg.ChannelType != ChannelDirect {
|
||||
return true
|
||||
}
|
||||
now := g.now()
|
||||
key := ConversationKey(msg)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
st := g.sessionLocked(key)
|
||||
st.directHits = g.trimHits(st.directHits, now)
|
||||
if len(st.directHits) >= g.directPerMin {
|
||||
g.auditMsg(msg, "rate_limited")
|
||||
return false
|
||||
}
|
||||
st.directHits = append(st.directHits, now)
|
||||
return true
|
||||
}
|
||||
|
||||
// AcquireTurn reserves the single running-turn slot of a conversation, returning
|
||||
// false when a turn is already in flight for it. Callers must pair every true
|
||||
// result with ReleaseTurn. It is the second half of the mandatory stranger flow
|
||||
// control (主人 2026-10-05 裁决③：「同一发送者同时最多 1 个在跑的回合」): a stranger
|
||||
// can send several messages inside the per-minute allowance, yet only one of them
|
||||
// may occupy an expensive agent turn at a time. It never decides identity — the
|
||||
// key is the same structural ConversationKey the rest of the gate uses.
|
||||
func (g *FlowGate) AcquireTurn(msg UnifiedMessage) bool {
|
||||
key := ConversationKey(msg)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
st := g.sessionLocked(key)
|
||||
if st.inFlight >= 1 {
|
||||
return false
|
||||
}
|
||||
st.inFlight++
|
||||
return true
|
||||
}
|
||||
|
||||
// ReleaseTurn releases a slot reserved by AcquireTurn.
|
||||
func (g *FlowGate) ReleaseTurn(msg UnifiedMessage) {
|
||||
key := ConversationKey(msg)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
st := g.sessionLocked(key)
|
||||
if st.inFlight > 0 {
|
||||
st.inFlight--
|
||||
}
|
||||
}
|
||||
|
||||
// drop builds a refused verdict from a single message that never entered the
|
||||
// buffer (rate-limited at the door), so the audit hook still sees it.
|
||||
func (g *FlowGate) drop(reason string, msg UnifiedMessage) GateVerdict {
|
||||
@@ -281,7 +359,8 @@ func (g *FlowGate) sessionLocked(key string) *flowSession {
|
||||
}
|
||||
|
||||
// maybeGCLocked 每 sessionGCInterval 次放行清理一次空闲会话，避免会话状态表随历史会话
|
||||
// 无界增长：滑窗时间戳全部过期且待处理队列为空的会话直接删除。调用方须持有 g.mu。
|
||||
// 无界增长：滑窗时间戳全部过期、待处理队列为空、且没有回合在跑的会话直接删除。
|
||||
// 调用方须持有 g.mu。
|
||||
func (g *FlowGate) maybeGCLocked(now time.Time) {
|
||||
g.admitted++
|
||||
if g.admitted%sessionGCInterval != 0 {
|
||||
@@ -289,7 +368,9 @@ func (g *FlowGate) maybeGCLocked(now time.Time) {
|
||||
}
|
||||
for key, st := range g.sessions {
|
||||
st.groupHits = g.trimHits(st.groupHits, now)
|
||||
if len(st.groupHits) == 0 && len(st.buffer) == 0 {
|
||||
st.directHits = g.trimHits(st.directHits, now)
|
||||
if len(st.groupHits) == 0 && len(st.directHits) == 0 &&
|
||||
len(st.buffer) == 0 && st.inFlight == 0 {
|
||||
delete(g.sessions, key)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
package adapter
|
||||
|
||||
// 本文件覆盖主人 2026-10-05 裁决③的**陌生人私聊流控**两件套：
|
||||
// - AllowDirect：私聊按发送者（会话键 platform+channel+sender）的每分钟滑窗；
|
||||
// - AcquireTurn/ReleaseTurn：同一发送者同时最多 1 个在跑的回合。
|
||||
// 两者都复用同一张 flowSession 表与同一套 ConversationKey 口径，不另造第二套限流。
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestFlowGate_AllowDirectIsPerSender 钉住私聊滑窗按发送者独立计数：
|
||||
// A 用尽额度不挤占 B；窗口滚过后 A 恢复；群聊消息不经这条通道。
|
||||
func TestFlowGate_AllowDirectIsPerSender(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
now, clock := fixedClock(time.Unix(2000, 0))
|
||||
var audited []string
|
||||
g := NewFlowGate(FlowGateOptions{
|
||||
DirectMsgPerMin: 2,
|
||||
Window: time.Minute,
|
||||
Now: now,
|
||||
OnPruned: func(_ UnifiedMessage, reason string) { audited = append(audited, reason) },
|
||||
})
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
if !g.AllowDirect(directMsgTo("peer-1", i)) {
|
||||
t.Fatalf("peer-1 第 %d 条被拒，want allowed", i+1)
|
||||
}
|
||||
}
|
||||
if g.AllowDirect(directMsgTo("peer-1", 2)) {
|
||||
t.Fatal("peer-1 超出自身额度仍被放行")
|
||||
}
|
||||
if len(audited) != 1 || audited[0] != "rate_limited" {
|
||||
t.Fatalf("拒绝未入审计：%v", audited)
|
||||
}
|
||||
if !g.AllowDirect(directMsgTo("peer-2", 0)) {
|
||||
t.Fatal("peer-2 被 peer-1 的额度挤占（私聊限流未按发送者独立）")
|
||||
}
|
||||
// 群聊走 Admit 的群聊滑窗，AllowDirect 对它恒放行。
|
||||
if !g.AllowDirect(groupMsgIn("-100", 0)) {
|
||||
t.Fatal("群聊消息不该被私聊限流拦下")
|
||||
}
|
||||
|
||||
*clock = clock.Add(61 * time.Second)
|
||||
if !g.AllowDirect(directMsgTo("peer-1", 3)) {
|
||||
t.Fatal("窗口滚过后 peer-1 应恢复放行")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowGate_AllowDirectDisabledByZero 钉住零值 = 关闭：没有配置私聊限流时
|
||||
// （DirectMsgPerMin 为 0）AllowDirect 恒放行，既有调用行为不变。
|
||||
func TestFlowGate_AllowDirectDisabledByZero(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
g := NewFlowGate(FlowGateOptions{})
|
||||
for i := 0; i < 50; i++ {
|
||||
if !g.AllowDirect(directMsgTo("peer-1", i)) {
|
||||
t.Fatalf("零值额度下第 %d 条被拒，want allowed", i+1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowDirectDefaultIsPinned 用**字面量 6** 钉住保护性默认值（与 FlowLimitDefault
|
||||
// 的既有钉桩同口径）。不这么钉的后果：`stranger_test.go` 的额度循环以该常量为上界，
|
||||
// 一旦有人把它改成近似无限，测试不会断言失败，而是**跑一百万轮把整包测试拖到超时**
|
||||
// ——看起来像「偶发卡死」，实际是流控被静默关掉。主人 2026-10-05 裁决③要求陌生人
|
||||
// 私聊**必须有**流控，因此这条默认值属于契约，不允许随手放大。
|
||||
func TestFlowDirectDefaultIsPinned(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if FlowDirectDefault != 6 {
|
||||
t.Fatalf("FlowDirectDefault = %d, want 6（每发送者每分钟 6 条的保护性默认；"+
|
||||
"放大它等于静默关闭陌生人私聊流控）", FlowDirectDefault)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowGate_AcquireTurnSinglePerConversation 钉住「同一会话同时最多 1 个在跑的回合」：
|
||||
// 第一个占位成功、第二个被拒；换一个发送者互不影响；释放后又能占。
|
||||
func TestFlowGate_AcquireTurnSinglePerConversation(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
g := NewFlowGate(FlowGateOptions{})
|
||||
a := directMsgTo("peer-1", 0)
|
||||
b := directMsgTo("peer-2", 0)
|
||||
|
||||
if !g.AcquireTurn(a) {
|
||||
t.Fatal("首个回合应能占到槽位")
|
||||
}
|
||||
if g.AcquireTurn(a) {
|
||||
t.Fatal("同一发送者的第二个回合不该同时占到槽位")
|
||||
}
|
||||
if !g.AcquireTurn(b) {
|
||||
t.Fatal("另一个发送者的槽位被挤占")
|
||||
}
|
||||
|
||||
g.ReleaseTurn(a)
|
||||
if !g.AcquireTurn(a) {
|
||||
t.Fatal("释放后应能重新占到槽位")
|
||||
}
|
||||
g.ReleaseTurn(a)
|
||||
g.ReleaseTurn(b)
|
||||
}
|
||||
Reference in new issue
Block a user