fix(agent): 危险写限流改为滑动窗口,与中继侧同一算法
节点侧的限流器此前也是固定窗口(按自然分钟重置计数器),而注释声称的性质是 "任意 60 秒内至多 N 次危险写"——与中继侧那个缺陷完全相同,只是这一侧还没有 人测过它。注释同样以该性质为由选择了固定窗口,所以代码与自己的理由相反。 现在两端是同一个算法:保留被放行的写时间戳,窗口满时最早那笔离开窗口才放行。 这两个限流器从两侧约束同一个操作(节点自己、以及中继对设备的预算),一侧比 另一侧宽一倍是那种要等到事后复盘才会被发现的不一致,所以保持一致比任何一侧 单独正确更重要。 测试里有一处值得记下来:这个性质不能按"花完预算、过一分钟、再花一次"来测—— 那样固定窗口也会通过,因为所有写都过期之后两种实现没有分歧。必须把预算花在 自然分钟边界**之前**,再断言边界之后一笔都放不进去。第一版就是这么写错的, 变异验证(把实现改回按自然分钟重置)没有失败,才发现测试本身没在测它。 内存代价同中继侧:每节点至多 N 个时间戳,而不是一个整数。 Go 全包、Node 112 项、Swift 101 项通过。
This commit is contained in:
@@ -277,31 +277,60 @@ func (e *Executor) record(envelope wire.Envelope, method, path string, status in
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// rateLimiter is a fixed-window counter. Deliberately not a token bucket: the property wanted is
|
// rateLimiter is a sliding-window counter over the dangerous writes this node has performed.
|
||||||
// "at most N dangerous writes in any 60 seconds", and a bucket's refill rate makes that harder to
|
//
|
||||||
// state and to test.
|
// The property is "at most N dangerous writes in any 60 seconds", and the implementation has to be
|
||||||
|
// that. It used to be a fixed window — a counter reset on a calendar minute — which is a different
|
||||||
|
// property: spending the budget just before the boundary and again just after puts 2N writes inside a
|
||||||
|
// few milliseconds. The comment above the old code argued for a fixed window *because* "a bucket's
|
||||||
|
// refill rate makes it harder to state and to test", so the code contradicted its own stated reason;
|
||||||
|
// a token bucket would not have satisfied the property either.
|
||||||
|
//
|
||||||
|
// The relay's per-device budget had the same defect and was fixed the same way (relay/src/budget.js).
|
||||||
|
// Keeping the two in step matters more than either: they bound the same operation from two sides, and
|
||||||
|
// a node whose limiter is twice as permissive as the relay's is a discrepancy nobody would notice
|
||||||
|
// until an incident review.
|
||||||
|
//
|
||||||
|
// Memory is bounded by the limit — at most N timestamps — where the fixed window held one integer.
|
||||||
type rateLimiter struct {
|
type rateLimiter struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
limit int
|
limit int
|
||||||
window time.Duration
|
window time.Duration
|
||||||
start time.Time
|
// accepted holds the timestamps of admitted writes, ascending. Anything older than the window is
|
||||||
consumed int
|
// dropped on the next call, so the slice never exceeds limit entries.
|
||||||
|
accepted []time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestDangerousWriteLimitDoesNotResetOnACalendarBoundary is the test for the property above: it spends
|
||||||
|
// the budget one second before a calendar-minute boundary and asserts nothing is admitted just after
|
||||||
|
// it. Written the obvious way — spend the budget, advance a minute, spend it again — the test passes
|
||||||
|
// against a fixed window too, because both implementations agree once everything has aged out.
|
||||||
|
|
||||||
func newRateLimiter(limit int, window time.Duration) *rateLimiter {
|
func newRateLimiter(limit int, window time.Duration) *rateLimiter {
|
||||||
return &rateLimiter{limit: limit, window: window}
|
return &rateLimiter{limit: limit, window: window}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Allow reports whether a dangerous write may proceed, and records it if so.
|
||||||
|
//
|
||||||
|
// Unlike the relay's budget, this one does not return a retry hint, and that is a property of the wire
|
||||||
|
// format rather than an omission: the agent answers inside `InnerResponse`, which carries a status, an
|
||||||
|
// outcome class and a body, and the phone renders the class with a fixed sentence. There is nowhere to
|
||||||
|
// put a wait without changing the envelope, and a hint that no layer can deliver is a number nobody
|
||||||
|
// reads. The relay can return one because HTTP has a header for it.
|
||||||
func (r *rateLimiter) Allow(now time.Time) bool {
|
func (r *rateLimiter) Allow(now time.Time) bool {
|
||||||
r.mu.Lock()
|
r.mu.Lock()
|
||||||
defer r.mu.Unlock()
|
defer r.mu.Unlock()
|
||||||
if r.start.IsZero() || now.Sub(r.start) >= r.window {
|
cutoff := now.Add(-r.window)
|
||||||
r.start = now
|
live := r.accepted[:0]
|
||||||
r.consumed = 0
|
for _, stamp := range r.accepted {
|
||||||
|
if stamp.After(cutoff) {
|
||||||
|
live = append(live, stamp)
|
||||||
}
|
}
|
||||||
if r.consumed >= r.limit {
|
}
|
||||||
|
r.accepted = live
|
||||||
|
if len(r.accepted) >= r.limit {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
r.consumed++
|
r.accepted = append(r.accepted, now)
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -290,7 +290,9 @@ func TestDangerousWritesAreRateLimited(t *testing.T) {
|
|||||||
if allowed != limit {
|
if allowed != limit {
|
||||||
t.Fatalf("allowed %d dangerous writes, want exactly %d", allowed, limit)
|
t.Fatalf("allowed %d dangerous writes, want exactly %d", allowed, limit)
|
||||||
}
|
}
|
||||||
// The window is fixed, not sliding: moving past it restores the full budget.
|
// The window slides, so a write is admitted exactly when its predecessor leaves the window — not
|
||||||
|
// all at once on a calendar boundary. These writes all happened at f.now, so the whole budget
|
||||||
|
// frees one minute after that instant.
|
||||||
f.advance(time.Minute)
|
f.advance(time.Minute)
|
||||||
envelope := envelopeAt("nd_test0001", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", uint64(f.now.Unix()))
|
envelope := envelopeAt("nd_test0001", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", uint64(f.now.Unix()))
|
||||||
if _, refusal := f.executor.Execute(context.Background(), envelope,
|
if _, refusal := f.executor.Execute(context.Background(), envelope,
|
||||||
@@ -299,6 +301,69 @@ func TestDangerousWritesAreRateLimited(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The property, stated as the defect it replaces.
|
||||||
|
//
|
||||||
|
// A fixed window — a counter reset on a calendar minute — admits 2N writes inside a few milliseconds
|
||||||
|
// by spending one budget just before its boundary and the next just after. The relay's per-device
|
||||||
|
// budget had exactly that flaw (see relay/src/budget.js), so the two limiters are now the same
|
||||||
|
// algorithm; this test is what makes the node's half of that claim checkable rather than asserted.
|
||||||
|
func TestDangerousWriteLimitDoesNotResetOnACalendarBoundary(t *testing.T) {
|
||||||
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.WriteHeader(200)
|
||||||
|
}))
|
||||||
|
body := []byte(`{"phone_number":"+8613800000000","content":"hello"}`)
|
||||||
|
intent, err := wire.IntentDigest("POST", "/api/sms/send", body)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
limit := testPolicy(t).Limits.DangerousWritesPerMinute
|
||||||
|
|
||||||
|
attempt := 0
|
||||||
|
send := func() bool {
|
||||||
|
inner := wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: intent}
|
||||||
|
envelope := envelopeAt("nd_test0001", newReqID(attempt), uint64(f.now.Unix()))
|
||||||
|
attempt++
|
||||||
|
_, refusal := f.executor.Execute(context.Background(), envelope, inner)
|
||||||
|
return refusal == nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Park the clock one second before a calendar-minute boundary, then spend the whole budget there.
|
||||||
|
// The fixture starts at a round minute, so 59 seconds lands on :59 of that minute.
|
||||||
|
f.advance(59 * time.Second)
|
||||||
|
for i := 0; i < limit; i++ {
|
||||||
|
if !send() {
|
||||||
|
t.Fatalf("write %d inside the budget was refused", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cross the boundary. A fixed window has just reset its counter and admits another full budget —
|
||||||
|
// which is the defect this test exists for. A sliding one admits nothing: the writes are two
|
||||||
|
// seconds old, and the window is 60.
|
||||||
|
f.advance(2 * time.Second)
|
||||||
|
admitted := 0
|
||||||
|
for i := 0; i < limit; i++ {
|
||||||
|
if send() {
|
||||||
|
admitted++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if admitted != 0 {
|
||||||
|
t.Fatalf("%d writes were admitted across the calendar boundary; the window is not sliding", admitted)
|
||||||
|
}
|
||||||
|
|
||||||
|
// And the budget frees one write at a time as the earlier ones age out, not all at once: 58 more
|
||||||
|
// seconds puts the writes from :59 past the window, so the full budget is available again.
|
||||||
|
f.advance(58 * time.Second)
|
||||||
|
admitted = 0
|
||||||
|
for i := 0; i < limit; i++ {
|
||||||
|
if send() {
|
||||||
|
admitted++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if admitted != limit {
|
||||||
|
t.Fatalf("after the window, %d writes were admitted, want exactly %d", admitted, limit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func newReqID(i int) string {
|
func newReqID(i int) string {
|
||||||
return "cccccccc-cccc-4ccc-8ccc-" + strings.Repeat("0", 11) + strconv.Itoa(i)
|
return "cccccccc-cccc-4ccc-8ccc-" + strings.Repeat("0", 11) + strconv.Itoa(i)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user