fix: 上行分发前统一限速,send 不再单独扣桶

This commit is contained in:
Nixevol
2026-09-30 16:21:05 +08:00
parent f139b9ed9b
commit aa24e974af
7 changed files with 155 additions and 13 deletions
+13
View File
@@ -107,6 +107,10 @@ func (u *appUplink) HandleUplink(ctx context.Context, conn port.ConnInfo, payloa
u.replyErr(ctx, conn, peekRID(payload), protocol.CodeBadRequest, err.Error()) u.replyErr(ctx, conn, peekRID(payload), protocol.CodeBadRequest, err.Error())
return nil return nil
} }
if !uplinkRateExempt(frame) && u.msg != nil && !u.msg.AllowRequest(conn.EndpointID) {
u.replyErr(ctx, conn, peekRID(payload), protocol.CodeRateLimited, "request rate exceeded")
return nil
}
rid, data, callErr := u.dispatch(ctx, conn, frame) rid, data, callErr := u.dispatch(ctx, conn, frame)
if callErr != nil { if callErr != nil {
@@ -117,6 +121,15 @@ func (u *appUplink) HandleUplink(ctx context.Context, conn port.ConnInfo, payloa
return nil return nil
} }
func uplinkRateExempt(frame any) bool {
switch frame.(type) {
case *protocol.Ack, *protocol.ReceiptAck:
return true
default:
return false
}
}
func (u *appUplink) dispatch(ctx context.Context, conn port.ConnInfo, frame any) (rid string, data any, err error) { func (u *appUplink) dispatch(ctx context.Context, conn port.ConnInfo, frame any) (rid string, data any, err error) {
switch f := frame.(type) { switch f := frame.(type) {
case *protocol.Send: case *protocol.Send:
+94
View File
@@ -0,0 +1,94 @@
package main
import (
"context"
"database/sql"
"io"
"log/slog"
"path/filepath"
"testing"
"time"
"git.asio.asia/nixevol/NixMsg/internal/app/message"
"git.asio.asia/nixevol/NixMsg/internal/app/port"
"git.asio.asia/nixevol/NixMsg/internal/auth"
"git.asio.asia/nixevol/NixMsg/internal/config"
"git.asio.asia/nixevol/NixMsg/internal/protocol"
"git.asio.asia/nixevol/NixMsg/internal/store"
)
func TestHandleUplinkRateLimitStatusAndAckExempt(t *testing.T) {
t.Parallel()
db, err := store.Open(filepath.Join(t.TempDir(), "data"), "FULL")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
nowMs := int64(1_700_000_000_000)
err = db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
_, e := tx.Exec(`
INSERT INTO endpoints(id, name, login_hash, talk_hash, talk_version, default_delay_ms, enabled, created_at)
VALUES('alice','alice','stub$login',NULL,0,0,1,?)`, nowMs)
return e
})
if err != nil {
t.Fatal(err)
}
lim := message.LimitsFromFullConfig(config.Default())
lim.RequestsPerSecond = 50
lim.RequestBurst = 100
app := message.New(db, lim, auth.NewStubHashPool(),
message.WithNow(func() time.Time { return time.UnixMilli(nowMs) }),
)
down := &message.RecordingDownlink{}
conns := message.NewMemoryConns()
conns.Set("alice", message.LiveConn{ConnID: "c1"})
u := &appUplink{msg: app, conns: conns, down: down, log: slog.New(slog.NewTextHandler(io.Discard, nil))}
conn := port.ConnInfo{EndpointID: "alice", ConnID: "c1"}
ctx := context.Background()
ackPayload, err := protocol.Marshal(&protocol.Ack{
V: protocol.Version, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "missing",
})
if err != nil {
t.Fatal(err)
}
for i := 0; i < 150; i++ {
if e := u.HandleUplink(ctx, conn, ackPayload); e != nil {
t.Fatal(e)
}
}
if n := countRespCode(down, protocol.CodeRateLimited); n != 0 {
t.Fatalf("ack should not count, rate_limited=%d", n)
}
statusPayload, err := protocol.Marshal(&protocol.Status{
V: protocol.Version, Type: protocol.TypeStatus, RID: "s", ID: "no-such",
})
if err != nil {
t.Fatal(err)
}
for i := 0; i < 150; i++ {
if e := u.HandleUplink(ctx, conn, statusPayload); e != nil {
t.Fatal(e)
}
}
limited := countRespCode(down, protocol.CodeRateLimited)
if limited != 50 {
t.Fatalf("status rate_limited=%d want 50 (burst 100 of 150)", limited)
}
}
func countRespCode(down *message.RecordingDownlink, code string) int {
n := 0
for _, p := range down.Snapshots() {
var resp protocol.Resp
if err := protocol.Unmarshal(p.Payload, &resp); err != nil {
continue
}
if !resp.OK && resp.Error != nil && resp.Error.Code == code {
n++
}
}
return n
}
+12 -3
View File
@@ -428,10 +428,10 @@
2. **请求频率突发容量写死为 100** 2. **请求频率突发容量写死为 100**
- 原条款:DEVELOPMENT 6.10 每端每秒 50、突发 100;配置示例仅有 `requests_per_second`。 - 原条款:DEVELOPMENT 6.10 每端每秒 50、突发 100;配置示例仅有 `requests_per_second`。
- 实际做法:`Limits.RequestBurst` 默认 100;`requests_per_second<=0` 时不限速(便于测试)。速率桶挂在 `message.App` 的 `Submit` 入口;`ack`/`receipt_ack` 不计入桶(与 6.10 一致)。 - 实际做法:`Limits.RequestBurst` 默认 100;`requests_per_second<=0` 时不限速(便于测试)。`message.App.AllowRequest` 导出同一令牌桶;`HandleUplink` 在分发前对 ack/receipt_ack 以外的帧调用。`Submit` 不再单独扣桶,避免 send 计两次。
- 原因:配置无独立 burst 字段。 - 原因:配置无独立 burst 字段。
- 备选方案:配置增加 `request_burst`;由连接线在上行统一限流。 - 备选方案:配置增加 `request_burst`。
- 影响:改 `requests_per_second` 不改突发;正式接线后若 N 线也限流可能双重计数。 - 影响:改 `requests_per_second` 不改突发;非 send 请求也受同一桶限制。
3. **未接线 `cmd/nixmsg`** 3. **未接线 `cmd/nixmsg`**
- 原条款:可替换 T0.4 假实现。 - 原条款:可替换 T0.4 假实现。
@@ -500,6 +500,15 @@
- 备选方案:在 group/identity 各自补写回执与收尾(继续分叉)。 - 备选方案:在 group/identity 各自补写回执与收尾(继续分叉)。
- 影响:退群/解散/停用后发送方可收到 rejected 回执,配额释放,正文删除。 - 影响:退群/解散/停用后发送方可收到 rejected 回执,配额释放,正文删除。
### 复审修复 C-05
1. **每端请求限速覆盖非 send 帧**
- 原条款:PRD F05 / DEVELOPMENT 6.10:除 ack、receipt_ack 外共用一个桶,默认每秒 50、突发 100。
- 实际做法:message 导出 `AllowRequest`;`cmd/nixmsg/uplink.go` 的 `HandleUplink` 解码后、分发前检查;超限回 `rate_limited`。去掉 `Submit` 内扣桶。不改 uplink 生命周期与 `publishResp`。
- 原因:原先只有 send 限速,unlock/status/目录/群等可打满哈希池与读库。
- 备选方案:把桶挪到 broker 层(B-09 范围)。
- 影响:开放注册后的非 send 请求也计入配额;直接调 `Submit` 的单测不再覆盖限速。
## 身份 I ## 身份 I
### I1 2026-09-30 ### I1 2026-09-30
+8
View File
@@ -57,3 +57,11 @@ func (r *rateLimiter) allow(endpointID string, now time.Time) bool {
b.tokens-- b.tokens--
return true return true
} }
// AllowRequest 消耗该端 1 个请求令牌;允许则 true。rps<=0 时不限速。
func (a *App) AllowRequest(endpointID string) bool {
if a == nil {
return true
}
return a.rates.allow(endpointID, a.now())
}
+22
View File
@@ -0,0 +1,22 @@
package message
import (
"testing"
)
func TestAllowRequestBurstAndAckExemptBucket(t *testing.T) {
t.Parallel()
lim := defaultTestLimits()
lim.RequestsPerSecond = 50
lim.RequestBurst = 100
app, _ := openTestApp(t, lim)
allowed := 0
for i := 0; i < 150; i++ {
if app.AllowRequest("alice") {
allowed++
}
}
if allowed != 100 {
t.Fatalf("allowed=%d want 100 (burst)", allowed)
}
}
-3
View File
@@ -21,9 +21,6 @@ func (a *App) Submit(ctx context.Context, senderID string, conn port.ConnInfo, r
return SubmitResult{}, errCode(protocol.CodeBadRequest, "invalid sender") return SubmitResult{}, errCode(protocol.CodeBadRequest, "invalid sender")
} }
now := a.now() now := a.now()
if !a.rates.allow(senderID, now) {
return SubmitResult{}, errCode(protocol.CodeRateLimited, "request rate exceeded")
}
if err := req.Validate(a.protocolLimits()); err != nil { if err := req.Validate(a.protocolLimits()); err != nil {
return SubmitResult{}, err return SubmitResult{}, err
+6 -7
View File
@@ -371,16 +371,15 @@ SELECT kind FROM talk_grants WHERE sender_id=? AND target_id=?`, "bob", "alice")
app, db := openTestApp(t, lim) app, db := openTestApp(t, lim)
insertEndpoint(t, db, "alice", "", 1, 0) insertEndpoint(t, db, "alice", "", 1, 0)
insertEndpoint(t, db, "bob", "", 1, 0) insertEndpoint(t, db, "bob", "", 1, 0)
if !app.AllowRequest("alice") || !app.AllowRequest("alice") {
t.Fatal("burst should allow first two")
}
if app.AllowRequest("alice") {
t.Fatal("third request should be rate limited")
}
ctx := context.Background() ctx := context.Background()
if _, err := app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("r1", "bob")); err != nil { if _, err := app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("r1", "bob")); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if _, err := app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("r2", "bob")); err != nil {
t.Fatal(err)
}
_, err := app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("r3", "bob"))
if protoCode(err) != protocol.CodeRateLimited {
t.Fatalf("want rate_limited got %v", err)
}
}) })
} }