From 3a9eec5de227dd4825fb80ab6019d872ad398871 Mon Sep 17 00:00:00 2001 From: chick Date: Thu, 17 Sep 2026 01:46:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E5=8D=B1=E9=99=A9=E5=86=99?= =?UTF-8?q?=E9=99=90=E6=B5=81=E6=94=B9=E4=B8=BA=E6=BB=91=E5=8A=A8=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=EF=BC=8C=E4=B8=8E=E4=B8=AD=E7=BB=A7=E4=BE=A7=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E7=AE=97=E6=B3=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 节点侧的限流器此前也是固定窗口（按自然分钟重置计数器），而注释声称的性质是 "任意 60 秒内至多 N 次危险写"——与中继侧那个缺陷完全相同，只是这一侧还没有 人测过它。注释同样以该性质为由选择了固定窗口，所以代码与自己的理由相反。 现在两端是同一个算法：保留被放行的写时间戳，窗口满时最早那笔离开窗口才放行。 这两个限流器从两侧约束同一个操作（节点自己、以及中继对设备的预算），一侧比 另一侧宽一倍是那种要等到事后复盘才会被发现的不一致，所以保持一致比任何一侧 单独正确更重要。 测试里有一处值得记下来：这个性质不能按"花完预算、过一分钟、再花一次"来测—— 那样固定窗口也会通过，因为所有写都过期之后两种实现没有分歧。必须把预算花在 自然分钟边界**之前**，再断言边界之后一笔都放不进去。第一版就是这么写错的， 变异验证（把实现改回按自然分钟重置）没有失败，才发现测试本身没在测它。 内存代价同中继侧：每节点至多 N 个时间戳，而不是一个整数。 Go 全包、Node 112 项、Swift 101 项通过。 --- internal/runtime/executor.go | 55 +++++++++++++++++++------- internal/runtime/runtime_test.go | 67 +++++++++++++++++++++++++++++++- 2 files changed, 108 insertions(+), 14 deletions(-) diff --git a/internal/runtime/executor.go b/internal/runtime/executor.go index 3861148..1f6bcd5 100644 --- a/internal/runtime/executor.go +++ b/internal/runtime/executor.go @@ -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 -// "at most N dangerous writes in any 60 seconds", and a bucket's refill rate makes that harder to -// state and to test. +// rateLimiter is a sliding-window counter over the dangerous writes this node has performed. +// +// 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 { - mu sync.Mutex - limit int - window time.Duration - start time.Time - consumed int + mu sync.Mutex + limit int + window time.Duration + // accepted holds the timestamps of admitted writes, ascending. Anything older than the window is + // 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 { 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 { r.mu.Lock() defer r.mu.Unlock() - if r.start.IsZero() || now.Sub(r.start) >= r.window { - r.start = now - r.consumed = 0 + cutoff := now.Add(-r.window) + live := r.accepted[: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 } - r.consumed++ + r.accepted = append(r.accepted, now) return true } diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index 706a77b..878d216 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -290,7 +290,9 @@ func TestDangerousWritesAreRateLimited(t *testing.T) { if 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) envelope := envelopeAt("nd_test0001", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", uint64(f.now.Unix())) 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 { return "cccccccc-cccc-4ccc-8ccc-" + strings.Repeat("0", 11) + strconv.Itoa(i) }