diff --git a/flake.nix b/flake.nix index 0c24413..c79091a 100644 --- a/flake.nix +++ b/flake.nix @@ -17,7 +17,7 @@ let lazuli = pkgs.buildGoModule rec { name = "lazuli"; - version = "0.1.0"; + version = "0.1.1"; src = pkgs.nix-gitignore.gitignoreSource [ "*.csv" "*.zip" "*.json" ] ./.; vendorHash = "sha256-Zr9gGytJARMbf/7120HYkKsfzpeW47MkwdMODD9QTKc="; ldflags = [ diff --git a/sync/rate.go b/sync/rate.go index 322f9b0..105717b 100644 --- a/sync/rate.go +++ b/sync/rate.go @@ -37,10 +37,19 @@ type KVStore interface { Set(key string, val int) error } +type Clock interface { + Now() time.Time +} + +type realClock struct{} + +func (realClock) Now() time.Time { return time.Now().UTC() } + type quotaLimiter struct { mu sync.Mutex kv KVStore prefix string + clock Clock } func (l *quotaLimiter) Stats() (int, int) { @@ -54,11 +63,12 @@ func NewRateLimiter(kv KVStore) RateLimiter { return "aLimiter{ kv: kv, prefix: "quota", + clock: realClock{}, } } func (l *quotaLimiter) getKeys() (string, string) { - day := time.Now().UTC().Format("2006-01-02") + day := l.clock.Now().Format("2006-01-02") return fmt.Sprintf("%s:writes:%s", l.prefix, day), fmt.Sprintf("%s:global:%s", l.prefix, day) } @@ -96,46 +106,15 @@ func (l *quotaLimiter) wait(ctx context.Context, wKey, gKey string, wCost, gCost defer l.mu.Unlock() for { - now := time.Now().UTC() - midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) - elapsed := now.Sub(midnight).Seconds() - - currW := 0 - if wKey != "" { - currW, _ = l.kv.Get(wKey) - } - currG, _ := l.kv.Get(gKey) - - var waitW, waitG time.Duration - - // Clamp the 'effective' elapsed time to prevent late-day bursts. - // We allow a maximum credit of 1 minute worth of rate limit (burst window). - const maxCreditSeconds = 60.0 - - if wLimit > 0 { - targetW := float64(currW+wCost) * secondsPerDay / float64(wLimit) - // effectively: we can only be 'ahead' by maxCreditSeconds - effectiveElapsed := max(elapsed, targetW-maxCreditSeconds) - if targetW > effectiveElapsed { - waitW = time.Duration((targetW - effectiveElapsed) * float64(time.Second)) - } - } - - if gLimit > 0 { - targetG := float64(currG+gCost) * secondsPerDay / float64(gLimit) - effectiveElapsed := max(elapsed, targetG-maxCreditSeconds) - if targetG > effectiveElapsed { - waitG = time.Duration((targetG - effectiveElapsed) * float64(time.Second)) - } - } + elapsed := l.getElapsedSinceMidnight() + currW, currG := l.getCurrentConsumption(wKey, gKey) + waitW := l.computeTargetWait(currW, wCost, wLimit, elapsed) + waitG := l.computeTargetWait(currG, gCost, gLimit, elapsed) maxWait := max(waitG, waitW) if maxWait <= 0 { - if wKey != "" { - l.kv.Set(wKey, currW+wCost) - } - l.kv.Set(gKey, currG+gCost) + l.updateConsumption(wKey, gKey, wCost, gCost, currW, currG) return nil } @@ -144,15 +123,63 @@ func (l *quotaLimiter) wait(ctx context.Context, wKey, gKey string, wCost, gCost } l.mu.Unlock() - timer := time.NewTimer(maxWait) - select { - case <-ctx.Done(): - timer.Stop() - l.mu.Lock() - return ctx.Err() - case <-timer.C: - l.mu.Lock() - continue + err := l.sleep(ctx, maxWait) + l.mu.Lock() + if err != nil { + return err } } } + +func (l *quotaLimiter) getElapsedSinceMidnight() float64 { + now := l.clock.Now() + midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) + return now.Sub(midnight).Seconds() +} + +func (l *quotaLimiter) getCurrentConsumption(wKey, gKey string) (int, int) { + currW := 0 + if wKey != "" { + currW, _ = l.kv.Get(wKey) + } + currG, _ := l.kv.Get(gKey) + return currW, currG +} + +func (l *quotaLimiter) computeTargetWait(curr, cost, limit int, elapsed float64) time.Duration { + if limit <= 0 { + return 0 + } + const maxCreditSeconds = 60.0 + target := float64(curr+cost) * secondsPerDay / float64(limit) + + // If the target time (when this consumption is 'earned') is within the + // burst window (now + maxCreditSeconds), we don't need to wait. + if target <= elapsed+maxCreditSeconds { + return 0 + } + + // Otherwise, we wait until we are at the edge of the burst window. + // We cap the wait at maxCreditSeconds to avoid overly long sleeps in a single loop, + // allowing for periodic re-checks of the clock and KV store. + wait := target - (elapsed + maxCreditSeconds) + return time.Duration(min(wait, maxCreditSeconds) * float64(time.Second)) +} + +func (l *quotaLimiter) updateConsumption(wKey, gKey string, wCost, gCost, currW, currG int) { + if wKey != "" { + l.kv.Set(wKey, currW+wCost) + } + l.kv.Set(gKey, currG+gCost) +} + +func (l *quotaLimiter) sleep(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/sync/rate_test.go b/sync/rate_test.go index fed265a..0f1abfc 100644 --- a/sync/rate_test.go +++ b/sync/rate_test.go @@ -63,64 +63,78 @@ func TestRateLimiter_Weighting(t *testing.T) { func TestRateLimiter_Smoothing(t *testing.T) { kv := &mockKV{data: make(map[string]int)} - limiter := NewRateLimiter(kv) - - // Set consumption to just below the allowed threshold for "now" - // Let's say 1 hour passed since midnight. - // Allowed = (9000 / 86400) * 3600 = 375 - - now := time.Now().UTC() - midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) - elapsed := now.Sub(midnight).Seconds() + clock := &mockClock{now: time.Date(2026, 1, 22, 1, 0, 0, 0, time.UTC)} + limiter := "aLimiter{ + kv: kv, + prefix: "quota", + clock: clock, + } - allowedW := int((float64(WriteLimitDay) / secondsPerDay) * elapsed) + // 1 hour since midnight = 3600s + // allowance = (9000/86400) * 3600 = 375 + // burst allowance = (9000/86400) * (3600 + 60) = 381.25 - wKey, _ := limiter.(*quotaLimiter).getKeys() - _ = kv.Set(wKey, allowedW) + _ = kv.Set("quota:writes:2026-01-22", 400) // Well over the limit + burst - // Trying to allow 100 more should block. - // We'll use a short timeout context to verify it blocks. ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) defer cancel() - err := limiter.AllowBulkWrite(ctx, 100) + err := limiter.AllowBulkWrite(ctx, 1) if err != context.DeadlineExceeded { t.Errorf("expected DeadlineExceeded, got %v", err) } } -func TestRateLimiter_Refund(t *testing.T) { - kv := &mockKV{data: make(map[string]int)} - limiter := NewRateLimiter(kv) - ctx := context.Background() +type mockClock struct { + now time.Time +} - // Initial state - w, g := limiter.Stats() - if w != 0 || g != 0 { - t.Errorf("expected clean start, got w=%d, g=%d", w, g) +func (m *mockClock) Now() time.Time { return m.now } + +func TestRateLimiter_MidnightRollover(t *testing.T) { + kv := &mockKV{data: make(map[string]int)} + clock := &mockClock{now: time.Date(2026, 1, 22, 23, 59, 59, 0, time.UTC)} + limiter := "aLimiter{ + kv: kv, + prefix: "quota", + clock: clock, } + ctx := context.Background() - // Consume + // 1. Consume some quota on day 1 err := limiter.AllowBulkWrite(ctx, 10) if err != nil { t.Fatal(err) } - w, g = limiter.Stats() - if w != 10 || g != 30 { - t.Errorf("expected w=10, g=30, got w=%d, g=%d", w, g) + + w1, g1 := limiter.Stats() + if w1 != 10 || g1 != 30 { + t.Errorf("expected w=10, g=30 on day 1, got w=%d, g=%d", w1, g1) } - // Refund half - limiter.RefundBulkWrite(5) - w, g = limiter.Stats() - if w != 5 || g != 15 { - t.Errorf("expected w=5, g=15, got w=%d, g=%d", w, g) + // 2. Advance time to next day + clock.now = clock.now.Add(2 * time.Second) // 00:00:01 on 2026-01-23 + + // 3. Stats should now reflect day 2 (0) + w2, g2 := limiter.Stats() + if w2 != 0 || g2 != 0 { + t.Errorf("expected w=0, g=0 on day 2, got w=%d, g=%d", w2, g2) } - // Refund more than exists (should floor at 0) - limiter.RefundBulkWrite(10) - w, g = limiter.Stats() - if w != 0 || g != 0 { - t.Errorf("expected floor at 0, got w=%d, g=%d", w, g) + // 4. Consumption on day 2 should not affect day 1 + err = limiter.AllowBulkWrite(ctx, 5) + if err != nil { + t.Fatal(err) + } + + w2, g2 = limiter.Stats() + if w2 != 5 || g2 != 15 { + t.Errorf("expected w=5, g=15 on day 2, got w=%d, g=%d", w2, g2) + } + + // Verify day 1 keys are still there but not accessed by Stats() + day1WKey := "quota:writes:2026-01-22" + if val, _ := kv.Get(day1WKey); val != 10 { + t.Errorf("expected day 1 write key to still be 10, got %d", val) } }