From a75318b811e8fd42a266101fa9d840d4a4c59394 Mon Sep 17 00:00:00 2001 From: rosemariejebbjtxbfp Date: Wed, 24 Jun 2026 06:52:51 +0000 Subject: [PATCH] feat(api): Redis read cache for product details and search Stage-0 caching from docs/scalability.md. The Go API now caches hot product-detail and search-result reads in Redis with a fail-open, epoch-versioned scheme; Python ingestion bumps the epoch after a write run to invalidate the cache globally in O(1). - api/internal/cache: fail-open Cache (GetJSON/SetJSON) namespaced by an epoch counter (og:cache:epoch); disabled when Redis is unconfigured. - store: ProductByID/ProductByGTIN (24h TTL) and SearchProducts (1h TTL) read-through the cache via WithCache. - ingestion: bump_cache_epoch() called after update_off/seed_off/ import_bypos/dedup commits when rows changed; best-effort, never fails a run. Adds redis dependency. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- api/cmd/server/main.go | 8 +- api/internal/cache/cache.go | 126 +++++++++++++++++++++++ api/internal/cache/cache_test.go | 84 +++++++++++++++ api/internal/store/store.go | 80 +++++++++++++- ingestion/opengoods/cache.py | 52 ++++++++++ ingestion/opengoods/jobs/dedup.py | 3 + ingestion/opengoods/jobs/import_bypos.py | 3 + ingestion/opengoods/jobs/seed_off.py | 3 + ingestion/opengoods/jobs/update_off.py | 3 + ingestion/pyproject.toml | 1 + ingestion/tests/test_cache.py | 44 ++++++++ 11 files changed, 403 insertions(+), 4 deletions(-) create mode 100644 api/internal/cache/cache.go create mode 100644 api/internal/cache/cache_test.go create mode 100644 ingestion/opengoods/cache.py create mode 100644 ingestion/tests/test_cache.py diff --git a/api/cmd/server/main.go b/api/cmd/server/main.go index 0b51902..5978d57 100644 --- a/api/cmd/server/main.go +++ b/api/cmd/server/main.go @@ -9,6 +9,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" + "github.com/baicai2026-baicai/goods/api/internal/cache" "github.com/baicai2026-baicai/goods/api/internal/config" "github.com/baicai2026-baicai/goods/api/internal/handler" "github.com/baicai2026-baicai/goods/api/internal/publicweb" @@ -36,7 +37,12 @@ func main() { if !limiter.Enabled() { log.Print("warning: Redis not configured; public API rate limiting disabled") } - h := handler.New(store.New(pool), publicweb.Dist()). + + readCache := cache.New(cfg.RedisURL) + if !readCache.Enabled() { + log.Print("warning: Redis not configured; public API read cache disabled") + } + h := handler.New(store.New(pool).WithCache(readCache), publicweb.Dist()). WithRateLimit(limiter, cfg.AnonRateLimitPerMin). WithQuotas(cfg.AnonTotalQuota, cfg.RegisteredRateLimitPerMin, cfg.RegisteredQuotaTotal) diff --git a/api/internal/cache/cache.go b/api/internal/cache/cache.go new file mode 100644 index 0000000..b92ad7c --- /dev/null +++ b/api/internal/cache/cache.go @@ -0,0 +1,126 @@ +// Package cache is a Redis-backed, fail-open read cache for the public API. +// +// It caches hot product details and search results so repeated reads avoid +// PostgreSQL. Like the ratelimit package, every operation fails open: if Redis +// is unavailable or misconfigured the caller simply falls back to the database, +// so the cache can never take the API down or serve stale data after Redis loss. +// +// Invalidation is global and O(1): keys are namespaced by an epoch counter +// stored in Redis (og:cache:epoch). The Python ingestion bumps that counter +// after a write run, which logically invalidates every cached entry at once +// while old keys age out via their TTL. The epoch is read at most once per +// refresh interval per process, so it adds no per-request round trip. +package cache + +import ( + "context" + "encoding/json" + "errors" + "log" + "strconv" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// epochKey is the Redis key holding the global cache generation counter. +const epochKey = "og:cache:epoch" + +// epochRefresh bounds how often a process re-reads the epoch from Redis. +const epochRefresh = 10 * time.Second + +// opTimeout caps any single Redis operation so a slow backend never blocks a +// request beyond this; on timeout the cache fails open. +const opTimeout = 150 * time.Millisecond + +// Cache wraps a Redis client. A nil-backed Cache (Redis unconfigured) disables +// caching: every Get misses and every Set is a no-op. +type Cache struct { + rdb *redis.Client + + mu sync.RWMutex + epoch int64 + epochSetAt time.Time + epochOK bool +} + +// New builds a Cache from a redis:// URL. On a parse error it logs and returns a +// disabled (fail-open) cache so the server still boots. +func New(redisURL string) *Cache { + opt, err := redis.ParseURL(redisURL) + if err != nil { + log.Printf("cache: invalid redis url %q: %v (caching disabled)", redisURL, err) + return &Cache{} + } + return &Cache{rdb: redis.NewClient(opt)} +} + +// Enabled reports whether a Redis backend is configured. +func (c *Cache) Enabled() bool { return c != nil && c.rdb != nil } + +// epochNow returns the current cache generation, reading it from Redis at most +// once per epochRefresh. On any Redis error it keeps the last known value and +// throttles re-reads so a down backend cannot slow the hot path. +func (c *Cache) epochNow(ctx context.Context) int64 { + c.mu.RLock() + if c.epochOK && time.Since(c.epochSetAt) < epochRefresh { + e := c.epoch + c.mu.RUnlock() + return e + } + c.mu.RUnlock() + + cctx, cancel := context.WithTimeout(ctx, opTimeout) + defer cancel() + n, err := c.rdb.Get(cctx, epochKey).Int64() + + c.mu.Lock() + defer c.mu.Unlock() + switch { + case err == nil: + c.epoch = n + case errors.Is(err, redis.Nil): + c.epoch = 0 + } + c.epochSetAt = time.Now() + c.epochOK = true + return c.epoch +} + +// key namespaces a logical suffix under the current epoch. +func (c *Cache) key(ctx context.Context, suffix string) string { + return "og:v" + strconv.FormatInt(c.epochNow(ctx), 10) + ":" + suffix +} + +// GetJSON unmarshals the cached value for suffix into dest and reports a hit. +// Any miss, decode error, or Redis error returns false (fail-open). +func (c *Cache) GetJSON(ctx context.Context, suffix string, dest any) bool { + if !c.Enabled() { + return false + } + k := c.key(ctx, suffix) + cctx, cancel := context.WithTimeout(ctx, opTimeout) + defer cancel() + b, err := c.rdb.Get(cctx, k).Bytes() + if err != nil { + return false + } + return json.Unmarshal(b, dest) == nil +} + +// SetJSON stores val (JSON-encoded) for suffix with the given TTL. Best effort: +// marshal or Redis errors are ignored. +func (c *Cache) SetJSON(ctx context.Context, suffix string, val any, ttl time.Duration) { + if !c.Enabled() { + return + } + b, err := json.Marshal(val) + if err != nil { + return + } + k := c.key(ctx, suffix) + cctx, cancel := context.WithTimeout(ctx, opTimeout) + defer cancel() + _ = c.rdb.Set(cctx, k, b, ttl).Err() +} diff --git a/api/internal/cache/cache_test.go b/api/internal/cache/cache_test.go new file mode 100644 index 0000000..15e6d18 --- /dev/null +++ b/api/internal/cache/cache_test.go @@ -0,0 +1,84 @@ +package cache + +import ( + "context" + "fmt" + "os" + "testing" + "time" +) + +// TestDisabledFailsOpen verifies a Cache without a Redis backend never panics, +// always misses, and silently drops writes. +func TestDisabledFailsOpen(t *testing.T) { + c := New("not-a-valid-url") // parse error => disabled + if c.Enabled() { + t.Fatal("expected cache to be disabled for invalid url") + } + c.SetJSON(context.Background(), "k", map[string]int{"a": 1}, time.Minute) + var dst map[string]int + if c.GetJSON(context.Background(), "k", &dst) { + t.Fatalf("disabled cache must always miss, got %+v", dst) + } +} + +func testCache(t *testing.T) *Cache { + t.Helper() + url := os.Getenv("OPENGOODS_REDIS_URL") + if url == "" { + url = "redis://localhost:6379/0" + } + c := New(url) + if !c.Enabled() { + t.Skip("redis not configured") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := c.rdb.Ping(ctx).Err(); err != nil { + t.Skipf("redis not reachable: %v", err) + } + return c +} + +// TestRoundTrip stores then reads a value back. +func TestRoundTrip(t *testing.T) { + c := testCache(t) + ctx := context.Background() + suffix := fmt.Sprintf("test:rt:%d", time.Now().UnixNano()) + + c.SetJSON(ctx, suffix, map[string]any{"name": "foo", "n": float64(3)}, time.Minute) + got := map[string]any{} + if !c.GetJSON(ctx, suffix, &got) { + t.Fatal("expected cache hit after set") + } + if got["name"] != "foo" || got["n"] != float64(3) { + t.Fatalf("unexpected payload: %+v", got) + } +} + +// TestEpochInvalidation verifies that bumping the epoch counter logically drops +// every previously cached entry. +func TestEpochInvalidation(t *testing.T) { + c := testCache(t) + ctx := context.Background() + suffix := fmt.Sprintf("test:epoch:%d", time.Now().UnixNano()) + + c.SetJSON(ctx, suffix, map[string]int{"v": 1}, time.Minute) + var dst map[string]int + if !c.GetJSON(ctx, suffix, &dst) { + t.Fatal("expected hit before epoch bump") + } + + // Simulate an ingestion write bumping the global epoch. + if err := c.rdb.Incr(ctx, epochKey).Err(); err != nil { + t.Fatalf("incr epoch: %v", err) + } + // Force the process to re-read the epoch rather than use its cached value. + c.mu.Lock() + c.epochOK = false + c.mu.Unlock() + + if c.GetJSON(ctx, suffix, &dst) { + t.Fatal("entry should be invisible after epoch bump") + } +} diff --git a/api/internal/store/store.go b/api/internal/store/store.go index b494e47..f9bafad 100644 --- a/api/internal/store/store.go +++ b/api/internal/store/store.go @@ -4,21 +4,35 @@ package store import ( "context" + "crypto/sha1" + "encoding/hex" "encoding/json" "errors" "strconv" "strings" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + + "github.com/baicai2026-baicai/goods/api/internal/cache" +) + +// Cache TTLs for the public read cache. Product details are far less volatile +// than search result sets, so they live longer; both are also invalidated +// wholesale whenever ingestion bumps the cache epoch. +const ( + productCacheTTL = 24 * time.Hour + searchCacheTTL = time.Hour ) // ErrNotFound is returned when a requested row does not exist. var ErrNotFound = errors.New("not found") -// Store wraps a PostgreSQL connection pool. +// Store wraps a PostgreSQL connection pool and an optional read cache. type Store struct { - pool *pgxpool.Pool + pool *pgxpool.Pool + cache *cache.Cache } // New constructs a Store from an existing pgx pool. @@ -26,6 +40,30 @@ func New(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } +// WithCache attaches a Redis-backed read cache. A nil or disabled cache leaves +// the Store reading straight from PostgreSQL. +func (s *Store) WithCache(c *cache.Cache) *Store { + s.cache = c + return s +} + +// cacheGet reads a cached JSON value into dest, reporting a hit. It is a no-op +// miss when no cache is attached. +func (s *Store) cacheGet(ctx context.Context, suffix string, dest any) bool { + if s.cache == nil { + return false + } + return s.cache.GetJSON(ctx, suffix, dest) +} + +// cacheSet stores a JSON value when a cache is attached. +func (s *Store) cacheSet(ctx context.Context, suffix string, val any, ttl time.Duration) { + if s.cache == nil { + return + } + s.cache.SetJSON(ctx, suffix, val, ttl) +} + // Ping verifies database connectivity. func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) @@ -222,6 +260,10 @@ func stringifyAttr(v any) string { // ProductByGTIN looks up an active product by any of its barcodes. func (s *Store) ProductByGTIN(ctx context.Context, gtin string) (*Product, error) { + const suffix = "prod:gtin:" + if cached := new(Product); s.cacheGet(ctx, suffix+gtin, cached) { + return cached, nil + } row := s.pool.QueryRow(ctx, productSelect+` WHERE p.status = 'active' AND (p.gtin = $1 OR EXISTS ( @@ -238,11 +280,16 @@ func (s *Store) ProductByGTIN(ctx context.Context, gtin string) (*Product, error if p.Specs, err = s.buildSpecs(ctx, p.ArchiveKind, attrs); err != nil { return nil, err } + s.cacheSet(ctx, suffix+gtin, p, productCacheTTL) return p, nil } // ProductByID looks up a product by its UUID. func (s *Store) ProductByID(ctx context.Context, id string) (*Product, error) { + const suffix = "prod:id:" + if cached := new(Product); s.cacheGet(ctx, suffix+id, cached) { + return cached, nil + } row := s.pool.QueryRow(ctx, productSelect+" WHERE p.id = $1", id) p, attrs, err := scanProduct(row) if err != nil { @@ -254,6 +301,7 @@ func (s *Store) ProductByID(ctx context.Context, id string) (*Product, error) { if p.Specs, err = s.buildSpecs(ctx, p.ArchiveKind, attrs); err != nil { return nil, err } + s.cacheSet(ctx, suffix+id, p, productCacheTTL) return p, nil } @@ -268,6 +316,11 @@ const fuzzyThreshold = "0.42" // similarity blended with quality_score so the best, most-complete records // surface first. Without a query, results are ordered by quality_score. func (s *Store) SearchProducts(ctx context.Context, f SearchFilters, limit, offset int) ([]ProductSummary, int, error) { + suffix := searchCacheSuffix(f, limit, offset) + if entry := new(searchCacheEntry); s.cacheGet(ctx, suffix, entry) { + return entry.Items, entry.Total, nil + } + args := []any{} where := "WHERE p.status = 'active'" @@ -334,7 +387,28 @@ LEFT JOIN category c ON c.id = p.category_id ` } out = append(out, ps) } - return out, total, rows.Err() + if err := rows.Err(); err != nil { + return nil, 0, err + } + s.cacheSet(ctx, suffix, searchCacheEntry{Items: out, Total: total}, searchCacheTTL) + return out, total, nil +} + +// searchCacheEntry is the cached payload for a SearchProducts call. +type searchCacheEntry struct { + Items []ProductSummary `json:"items"` + Total int `json:"total"` +} + +// searchCacheSuffix derives a stable cache key from the full filter set and +// paging window so distinct queries never collide. +func searchCacheSuffix(f SearchFilters, limit, offset int) string { + raw := strings.Join([]string{ + f.Query, f.Category, f.Brand, f.Country, + strconv.Itoa(limit), strconv.Itoa(offset), + }, "\x1f") + sum := sha1.Sum([]byte(raw)) + return "search:" + hex.EncodeToString(sum[:]) } // Nutriments returns just the nutrition payload for a product. diff --git a/ingestion/opengoods/cache.py b/ingestion/opengoods/cache.py new file mode 100644 index 0000000..915ad2d --- /dev/null +++ b/ingestion/opengoods/cache.py @@ -0,0 +1,52 @@ +"""Read-cache invalidation signal for the public API. + +The Go API caches hot product details and search results in Redis, namespacing +every key by a global generation counter (``og:cache:epoch``). Bumping that +counter logically invalidates the entire cache in one O(1) operation while the +old keys age out via their TTL. + +Ingestion is the only writer to the database, so after a run that changed data +it calls :func:`bump_cache_epoch` to make those changes visible immediately +instead of waiting for per-key TTLs to expire. + +Like the API's cache, this is strictly best effort and fails open: if Redis is +unconfigured or unreachable the ingestion run still succeeds, and stale entries +simply expire on their own. +""" + +from __future__ import annotations + +import logging +import os + +logger = logging.getLogger(__name__) + +EPOCH_KEY = "og:cache:epoch" + + +def default_redis_url() -> str | None: + """Return the configured Redis URL, or ``None`` when caching is disabled.""" + return os.environ.get("OPENGOODS_REDIS_URL") or None + + +def bump_cache_epoch(url: str | None = None) -> bool: + """Increment the API cache generation counter. + + Returns ``True`` if the counter was bumped, ``False`` if caching is disabled + or Redis was unreachable. Never raises: invalidation failures must not fail + an ingestion run. + """ + url = url or default_redis_url() + if not url: + return False + try: + import redis # imported lazily so the dependency is optional at runtime + + client = redis.Redis.from_url(url, socket_timeout=2, socket_connect_timeout=2) + new_epoch = client.incr(EPOCH_KEY) + client.close() + logger.info("bumped API cache epoch to %s", new_epoch) + return True + except Exception as exc: # noqa: BLE001 - invalidation is best effort + logger.warning("cache epoch bump skipped: %s", exc) + return False diff --git a/ingestion/opengoods/jobs/dedup.py b/ingestion/opengoods/jobs/dedup.py index fc74713..eceb793 100644 --- a/ingestion/opengoods/jobs/dedup.py +++ b/ingestion/opengoods/jobs/dedup.py @@ -12,6 +12,7 @@ import sys import psycopg +from opengoods.cache import bump_cache_epoch from opengoods.etl.dedup import dedup_all from opengoods.etl.load import default_dsn @@ -23,6 +24,8 @@ def run(args: argparse.Namespace) -> int: conn.rollback() else: conn.commit() + if not args.dry_run and summary["merged"]: + bump_cache_epoch() mode = "dry-run" if args.dry_run else "applied" print(f"{mode} groups={summary['groups']} merged={summary['merged']}") return 0 diff --git a/ingestion/opengoods/jobs/import_bypos.py b/ingestion/opengoods/jobs/import_bypos.py index d466955..f875252 100644 --- a/ingestion/opengoods/jobs/import_bypos.py +++ b/ingestion/opengoods/jobs/import_bypos.py @@ -25,6 +25,7 @@ from collections.abc import Iterator import psycopg from opengoods.adapters.bypos import transform_bypos +from opengoods.cache import bump_cache_epoch from opengoods.etl.load import default_dsn, ensure_bypos_source, load_bypos_record_safe @@ -59,6 +60,8 @@ def run(args: argparse.Namespace) -> int: else: errored += 1 conn.commit() + if loaded: + bump_cache_epoch() print(f"loaded={loaded} skipped={skipped} errored={errored}") return 0 diff --git a/ingestion/opengoods/jobs/seed_off.py b/ingestion/opengoods/jobs/seed_off.py index f7d632e..06b924f 100644 --- a/ingestion/opengoods/jobs/seed_off.py +++ b/ingestion/opengoods/jobs/seed_off.py @@ -28,6 +28,7 @@ from opengoods.adapters.openfoodfacts import ( is_cn_gs1, read_dump, ) +from opengoods.cache import bump_cache_epoch from opengoods.etl.load import default_dsn, ensure_source, load_record_safe from opengoods.etl.transform import transform @@ -67,6 +68,8 @@ def run(args: argparse.Namespace) -> int: else: errored += 1 conn.commit() + if loaded: + bump_cache_epoch() print(f"loaded={loaded} skipped={skipped} errored={errored}") return 0 diff --git a/ingestion/opengoods/jobs/update_off.py b/ingestion/opengoods/jobs/update_off.py index ce67495..dfc507a 100644 --- a/ingestion/opengoods/jobs/update_off.py +++ b/ingestion/opengoods/jobs/update_off.py @@ -17,6 +17,7 @@ import sys import psycopg from opengoods.adapters.openfoodfacts import SOURCE_NAME, OpenFoodFactsAdapter +from opengoods.cache import bump_cache_epoch from opengoods.etl.load import default_dsn, ensure_source, load_record_safe from opengoods.etl.state import get_watermark, set_watermark from opengoods.etl.transform import transform @@ -49,6 +50,8 @@ def run(args: argparse.Namespace) -> int: stats={"loaded": loaded, "skipped": skipped, "errored": errored, "since": since}, ) conn.commit() + if loaded: + bump_cache_epoch() print( f"since={since} loaded={loaded} skipped={skipped} " f"errored={errored} watermark={high_watermark}" diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index 0cefe1c..1e88a81 100644 --- a/ingestion/pyproject.toml +++ b/ingestion/pyproject.toml @@ -6,6 +6,7 @@ requires-python = ">=3.11" dependencies = [ "httpx>=0.27", "psycopg[binary]>=3.2", + "redis>=5.0,<6", ] [project.optional-dependencies] diff --git a/ingestion/tests/test_cache.py b/ingestion/tests/test_cache.py new file mode 100644 index 0000000..7b3abc2 --- /dev/null +++ b/ingestion/tests/test_cache.py @@ -0,0 +1,44 @@ +"""Tests for the best-effort API cache invalidation signal.""" + +from __future__ import annotations + +import sys + +import opengoods.cache as cache + + +def test_disabled_when_no_url(monkeypatch): + monkeypatch.delenv("OPENGOODS_REDIS_URL", raising=False) + assert cache.default_redis_url() is None + assert cache.bump_cache_epoch() is False + + +def test_bump_fails_open_when_unreachable(monkeypatch): + # An unroutable URL must not raise; the run still succeeds. + monkeypatch.setenv("OPENGOODS_REDIS_URL", "redis://127.0.0.1:1/0") + assert cache.bump_cache_epoch() is False + + +def test_bump_increments_epoch(monkeypatch): + calls = {} + + class FakeClient: + def incr(self, key): + calls["key"] = key + return 7 + + def close(self): + calls["closed"] = True + + class FakeRedis: + @staticmethod + def from_url(url, **kwargs): + calls["url"] = url + return FakeClient() + + monkeypatch.setitem(sys.modules, "redis", type("M", (), {"Redis": FakeRedis})) + + assert cache.bump_cache_epoch("redis://example:6379/0") is True + assert calls["key"] == "og:cache:epoch" + assert calls["url"] == "redis://example:6379/0" + assert calls["closed"] is True