perf(S01/T01): Add token-bucket rate limiter (1 req/sec), SQLite respon…
- backend/explore/ratelimiter.go - backend/explore/cache.go - backend/database/sql/schemas/explore_cache.sql - backend/database/database.go
This commit is contained in:
@@ -0,0 +1,123 @@
|
||||
package explore
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"yellowjacket/backend/database"
|
||||
)
|
||||
|
||||
// Cache provides a SQLite-backed response cache with TTL expiry.
|
||||
// It stores raw JSON API responses keyed by URL and supports
|
||||
// optional MBID columns for future autotagging lookups.
|
||||
//
|
||||
// All operations use the shared database.DB connection and its
|
||||
// single-writer constraint (SetMaxOpenConns(1)).
|
||||
type Cache struct {
|
||||
db *database.DB
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewCache returns a cache backed by the given database connection.
|
||||
func NewCache(db *database.DB, logger *slog.Logger) *Cache {
|
||||
return &Cache{db: db, logger: logger}
|
||||
}
|
||||
|
||||
// Get returns the cached response for the given URL key if it
|
||||
// exists and has not expired. Returns (data, true) on a cache hit
|
||||
// and (nil, false) on a miss or expired entry.
|
||||
func (c *Cache) Get(key string) ([]byte, bool) {
|
||||
rows, err := c.db.QueryContext(
|
||||
"SELECT response FROM explore_cache WHERE url_key = ? AND expires_at > datetime('now')",
|
||||
key,
|
||||
)
|
||||
if err != nil {
|
||||
c.logger.Warn("explore cache get error",
|
||||
"key", key,
|
||||
"err", err,
|
||||
)
|
||||
|
||||
return nil, false
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
if !rows.Next() {
|
||||
c.logger.Debug("explore cache miss", "key", key)
|
||||
|
||||
return nil, false
|
||||
}
|
||||
|
||||
var response string
|
||||
|
||||
if err := rows.Scan(&response); err != nil {
|
||||
c.logger.Warn("explore cache scan error",
|
||||
"key", key,
|
||||
"err", err,
|
||||
)
|
||||
|
||||
return nil, false
|
||||
}
|
||||
|
||||
c.logger.Debug("explore cache hit", "key", key)
|
||||
|
||||
return []byte(response), true
|
||||
}
|
||||
|
||||
// Set stores a response in the cache with the given TTL. If mbid
|
||||
// and entityType are non-empty they are stored for future
|
||||
// autotagging lookups; otherwise they are stored as NULL.
|
||||
func (c *Cache) Set(
|
||||
key string,
|
||||
data []byte,
|
||||
ttl time.Duration,
|
||||
mbid string,
|
||||
entityType string,
|
||||
) {
|
||||
seconds := int(ttl.Seconds())
|
||||
if seconds < 1 {
|
||||
seconds = 1
|
||||
}
|
||||
|
||||
expr := fmt.Sprintf("datetime('now', '+%d seconds')", seconds)
|
||||
|
||||
query := fmt.Sprintf(
|
||||
`INSERT OR REPLACE INTO explore_cache
|
||||
(url_key, response, mbid, entity_type, expires_at)
|
||||
VALUES (?, ?, NULLIF(?, ''), NULLIF(?, ''), %s)`,
|
||||
expr,
|
||||
)
|
||||
|
||||
if _, err := c.db.ExecContext(query, key, string(data), mbid, entityType); err != nil {
|
||||
c.logger.Warn("explore cache set error",
|
||||
"key", key,
|
||||
"err", err,
|
||||
)
|
||||
} else {
|
||||
c.logger.Debug("explore cache set",
|
||||
"key", key,
|
||||
"ttl", ttl,
|
||||
"mbid", mbid,
|
||||
"entityType", entityType,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Evict removes all expired entries from the cache.
|
||||
func (c *Cache) Evict() {
|
||||
result, err := c.db.ExecContext(
|
||||
"DELETE FROM explore_cache WHERE expires_at < datetime('now')",
|
||||
)
|
||||
if err != nil {
|
||||
c.logger.Warn("explore cache evict error", "err", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if n, _ := result.RowsAffected(); n > 0 {
|
||||
c.logger.Info("explore cache evicted expired entries",
|
||||
"count", n,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package explore
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"yellowjacket/backend/database"
|
||||
)
|
||||
|
||||
func newTestCache(t *testing.T) *Cache {
|
||||
t.Helper()
|
||||
|
||||
db := database.NewTestDB(t)
|
||||
|
||||
return NewCache(db, slog.Default())
|
||||
}
|
||||
|
||||
func TestCacheSetGet(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c := newTestCache(t)
|
||||
|
||||
data := []byte(`{"artist":"Radiohead"}`)
|
||||
c.Set("https://musicbrainz.org/ws/2/artist?query=radiohead", data, 5*time.Minute, "", "")
|
||||
|
||||
got, ok := c.Get("https://musicbrainz.org/ws/2/artist?query=radiohead")
|
||||
if !ok {
|
||||
t.Fatal("expected cache hit, got miss")
|
||||
}
|
||||
|
||||
if string(got) != string(data) {
|
||||
t.Errorf("got %q, want %q", string(got), string(data))
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheMiss(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c := newTestCache(t)
|
||||
|
||||
_, ok := c.Get("https://nonexistent.example.com/api")
|
||||
if ok {
|
||||
t.Error("expected cache miss, got hit")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheTTLExpiry(t *testing.T) {
|
||||
c := newTestCache(t)
|
||||
|
||||
data := []byte(`{"ephemeral":true}`)
|
||||
c.Set("ttl-test-key", data, 1*time.Second, "", "")
|
||||
|
||||
// Verify it's there immediately.
|
||||
if _, ok := c.Get("ttl-test-key"); !ok {
|
||||
t.Fatal("expected cache hit immediately after set")
|
||||
}
|
||||
|
||||
// Wait for expiry.
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
if _, ok := c.Get("ttl-test-key"); ok {
|
||||
t.Error("expected cache miss after TTL expiry, got hit")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheMBID(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c := newTestCache(t)
|
||||
|
||||
data := []byte(`{"name":"OK Computer"}`)
|
||||
c.Set(
|
||||
"mbid-test-key",
|
||||
data,
|
||||
10*time.Minute,
|
||||
"b3b40b1b-3c03-4b8a-8291-8e1f2d09e211",
|
||||
"release_group",
|
||||
)
|
||||
|
||||
// Query the MBID column directly to verify it was stored.
|
||||
db := c.db
|
||||
|
||||
rows, err := db.QueryContext(
|
||||
"SELECT mbid, entity_type FROM explore_cache WHERE url_key = ?",
|
||||
"mbid-test-key",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("query explore_cache: %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
if !rows.Next() {
|
||||
t.Fatal("explore_cache row not found")
|
||||
}
|
||||
|
||||
var (
|
||||
mbid sql.NullString
|
||||
entityType sql.NullString
|
||||
)
|
||||
|
||||
if err := rows.Scan(&mbid, &entityType); err != nil {
|
||||
t.Fatalf("scan: %v", err)
|
||||
}
|
||||
|
||||
if !mbid.Valid || mbid.String != "b3b40b1b-3c03-4b8a-8291-8e1f2d09e211" {
|
||||
t.Errorf("mbid = %v, want b3b40b1b-3c03-4b8a-8291-8e1f2d09e211", mbid)
|
||||
}
|
||||
|
||||
if !entityType.Valid || entityType.String != "release_group" {
|
||||
t.Errorf("entity_type = %v, want release_group", entityType)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCacheEvict(t *testing.T) {
|
||||
c := newTestCache(t)
|
||||
|
||||
// Insert an entry that expires in 1 second.
|
||||
c.Set("evict-key", []byte(`{}`), 1*time.Second, "", "")
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Evict expired entries.
|
||||
c.Evict()
|
||||
|
||||
// Verify the row is gone entirely (not just expired-but-present).
|
||||
db := c.db
|
||||
|
||||
rows, err := db.QueryContext(
|
||||
"SELECT COUNT(*) FROM explore_cache WHERE url_key = ?",
|
||||
"evict-key",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("query: %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
if !rows.Next() {
|
||||
t.Fatal("no row returned")
|
||||
}
|
||||
|
||||
var count int64
|
||||
if err := rows.Scan(&count); err != nil {
|
||||
t.Fatalf("scan: %v", err)
|
||||
}
|
||||
|
||||
if count != 0 {
|
||||
t.Errorf("expected 0 rows after evict, got %d", count)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
// Package explore provides MusicBrainz and ListenBrainz API clients
|
||||
// with rate-limited HTTP access and a SQLite response cache.
|
||||
package explore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// RateLimiter enforces a maximum request rate using a token bucket.
|
||||
// MusicBrainz requires ≤1 request per second and rejects ALL
|
||||
// requests (not just excess) when the rate is exceeded, so callers
|
||||
// block proactively via Wait rather than retrying reactively.
|
||||
//
|
||||
// RateLimiter is safe for concurrent use.
|
||||
type RateLimiter struct {
|
||||
limiter *rate.Limiter
|
||||
}
|
||||
|
||||
// NewRateLimiter returns a rate limiter that allows exactly one
|
||||
// request per second with a burst size of 1. The first call to
|
||||
// Wait returns immediately; subsequent calls block until the next
|
||||
// token is available.
|
||||
func NewRateLimiter() *RateLimiter {
|
||||
return &RateLimiter{
|
||||
limiter: rate.NewLimiter(rate.Every(time.Second), 1),
|
||||
}
|
||||
}
|
||||
|
||||
// Wait blocks until the rate limiter allows the caller to proceed
|
||||
// or the context is cancelled. Returns ctx.Err() if the context
|
||||
// expires before a token becomes available.
|
||||
func (r *RateLimiter) Wait(ctx context.Context) error {
|
||||
return r.limiter.Wait(ctx)
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package explore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRateLimiterBurst(t *testing.T) {
|
||||
rl := NewRateLimiter()
|
||||
ctx := context.Background()
|
||||
|
||||
const n = 5
|
||||
|
||||
start := time.Now()
|
||||
|
||||
for i := range n {
|
||||
if err := rl.Wait(ctx); err != nil {
|
||||
t.Fatalf("Wait %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
|
||||
// First request is immediate; 4 more at 1/sec = ≥4s total.
|
||||
if elapsed < 4*time.Second {
|
||||
t.Errorf(
|
||||
"elapsed %v, want ≥ 4s (rate limiter too fast)", elapsed,
|
||||
)
|
||||
}
|
||||
|
||||
// Generous upper bound to avoid CI flakes.
|
||||
if elapsed > 7*time.Second {
|
||||
t.Errorf(
|
||||
"elapsed %v, want ≤ 7s (rate limiter too slow)", elapsed,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimiterContextCancel(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rl := NewRateLimiter()
|
||||
|
||||
// Drain the initial token so the next Wait must block.
|
||||
if err := rl.Wait(context.Background()); err != nil {
|
||||
t.Fatalf("drain token: %v", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // cancel immediately
|
||||
|
||||
err := rl.Wait(ctx)
|
||||
if err == nil {
|
||||
t.Fatal("expected error from cancelled context, got nil")
|
||||
}
|
||||
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("error = %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user