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