Cache custom-query JSON responses to avoid API rate limits (#40)

* reporter: cache custom-query JSON responses (TTL + singleflight) to avoid price-source API rate limits

* cache: fix lint — gci import order + 'behaviour'->'behavior' (misspell)

Author: diorwave

custom_query/contracts/metrics/metrics.go

@@ -40,6 +40,18 @@ var (
 		Buckets: prometheus.DefBuckets,
 	})
 
+	// Response-cache metrics (see rpc_reader/cache.go). Useful to confirm the
+	// cache is actually shielding upstream APIs from rate limits.
+	RPCCacheHits = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "rpc_reader_cache_hits_total",
+		Help: "Total number of RPC responses served from the in-memory cache",
+	})
+
+	RPCCacheMisses = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "rpc_reader_cache_misses_total",
+		Help: "Total number of RPC cache misses that triggered an upstream fetch",
+	})
+
 	// Health check metrics (used by both contract and RPC readers)
 	RPCHealthCheckFailures = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "reader_health_check_failures_total",

custom_query/rpc/rpc_reader/cache.go

@@ -0,0 +1,151 @@
+package rpc_reader
+
+import (
+	"context"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/tellor-io/layer-daemons/custom_query/contracts/metrics"
+	"golang.org/x/sync/singleflight"
+)
+
+// The reporter polls the cyclelist roughly every 200ms, so a single custom
+// query can re-hit the same upstream JSON API (CoinGecko, CoinMarketCap, the
+// Uniswap/The Graph subgraph, osmosis, etc.) several times per second. With
+// free-tier API keys that quickly trips rate limits (HTTP 429) and produces the
+// "report generation failed" bursts.
+//
+// This cache sits in front of rpc_reader.FetchJSON and serves an identical
+// request from memory for up to CUSTOM_QUERY_CACHE_TTL, collapsing those
+// redundant calls into one upstream fetch per interval. Concurrent identical
+// requests are coalesced via singleflight so a cache miss can never cause a
+// stampede.
+//
+// Tuning (env vars, read once at startup):
+//   - CUSTOM_QUERY_CACHE_TTL : cache freshness window, a Go duration (e.g. "3s",
+//     "500ms"). Set to "0" to disable the cache entirely (pure pass-through to
+//     the previous behavior). Defaults to defaultCacheTTL.
+//
+// Correctness note: the TTL bounds how stale a reported price can be, so it must
+// stay well under the chain's value-freshness expectations. The default is
+// deliberately small.
+const (
+	cacheTTLEnv     = "CUSTOM_QUERY_CACHE_TTL"
+	defaultCacheTTL = 3 * time.Second
+)
+
+type cacheEntry struct {
+	body      []byte
+	fetchedAt time.Time
+}
+
+type responseCache struct {
+	ttl   time.Duration
+	mu    sync.RWMutex
+	items map[string]cacheEntry
+	group singleflight.Group
+}
+
+// sharedCache is process-wide: readers are rebuilt per query config, so the
+// cache must outlive any single Reader instance to be effective across cycles.
+var sharedCache = newResponseCache(cacheTTLFromEnv())
+
+func cacheTTLFromEnv() time.Duration {
+	v := strings.TrimSpace(os.Getenv(cacheTTLEnv))
+	if v == "" {
+		return defaultCacheTTL
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil || d < 0 {
+		return defaultCacheTTL
+	}
+	return d // 0 => disabled
+}
+
+func newResponseCache(ttl time.Duration) *responseCache {
+	return &responseCache{ttl: ttl, items: make(map[string]cacheEntry)}
+}
+
+func (c *responseCache) enabled() bool { return c.ttl > 0 }
+
+// get returns a cached body if present and still within the TTL.
+func (c *responseCache) get(key string) ([]byte, bool) {
+	c.mu.RLock()
+	e, ok := c.items[key]
+	c.mu.RUnlock()
+	if !ok || time.Since(e.fetchedAt) > c.ttl {
+		return nil, false
+	}
+	return e.body, true
+}
+
+func (c *responseCache) set(key string, body []byte) {
+	c.mu.Lock()
+	c.items[key] = cacheEntry{body: body, fetchedAt: time.Now()}
+	c.mu.Unlock()
+}
+
+// cacheKey uniquely identifies an HTTP request. It must capture everything that
+// can change the response: method, URL (which carries API-key query params),
+// POST body (GraphQL queries), and headers (which may carry API keys).
+func (r *Reader) cacheKey() string {
+	var b strings.Builder
+	b.WriteString(r.client.method)
+	b.WriteByte('\n')
+	b.WriteString(r.client.baseURL)
+	b.WriteByte('\n')
+	b.WriteString(r.Query)
+	b.WriteByte('\n')
+
+	keys := make([]string, 0, len(r.Headers))
+	for k := range r.Headers {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		b.WriteString(k)
+		b.WriteByte('=')
+		b.WriteString(r.Headers[k])
+		b.WriteByte('\n')
+	}
+	return b.String()
+}
+
+// FetchJSON returns the JSON response for the reader's request, served from the
+// in-memory cache when a fresh copy exists. When caching is disabled it falls
+// straight through to the underlying retrying fetch.
+//
+// The returned []byte may be shared with other callers; treat it as read-only
+// (all current callers only json.Unmarshal it, which does not mutate).
+func (r *Reader) FetchJSON(ctx context.Context) ([]byte, error) {
+	if !sharedCache.enabled() {
+		return r.fetchWithRetry(ctx)
+	}
+
+	key := r.cacheKey()
+	if body, ok := sharedCache.get(key); ok {
+		metrics.RPCCacheHits.Inc()
+		return body, nil
+	}
+	metrics.RPCCacheMisses.Inc()
+
+	v, err, _ := sharedCache.group.Do(key, func() (interface{}, error) {
+		// Re-check: a concurrent leader may have just populated the cache.
+		if body, ok := sharedCache.get(key); ok {
+			return body, nil
+		}
+		body, err := r.fetchWithRetry(ctx)
+		if err != nil {
+			return nil, err
+		}
+		sharedCache.set(key, body)
+		return body, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return v.([]byte), nil
+}

custom_query/rpc/rpc_reader/cache_test.go

@@ -0,0 +1,145 @@
+package rpc_reader
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// withSharedCache swaps the process-wide cache for the duration of a test.
+func withSharedCache(t *testing.T, ttl time.Duration) {
+	t.Helper()
+	prev := sharedCache
+	sharedCache = newResponseCache(ttl)
+	t.Cleanup(func() { sharedCache = prev })
+}
+
+// countingServer returns an httptest server that records how many times it was hit.
+func countingServer(t *testing.T, body string) (*httptest.Server, *int32) {
+	t.Helper()
+	var hits int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		atomic.AddInt32(&hits, 1)
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(srv.Close)
+	return srv, &hits
+}
+
+// newTestReader builds a reader with a generous timeout (NewReader treats the
+// timeout arg as seconds for the client and ms for the per-attempt context).
+func newTestReader(t *testing.T, url string) *Reader {
+	t.Helper()
+	r, err := NewReader(url, http.MethodGet, "", nil, nil, 5000, nil)
+	if err != nil {
+		t.Fatalf("NewReader: %v", err)
+	}
+	return r
+}
+
+func TestFetchJSON_ServesFromCacheWithinTTL(t *testing.T) {
+	withSharedCache(t, time.Minute)
+	srv, hits := countingServer(t, `{"price":1}`)
+	r := newTestReader(t, srv.URL)
+
+	for i := 0; i < 5; i++ {
+		body, err := r.FetchJSON(context.Background())
+		if err != nil {
+			t.Fatalf("FetchJSON: %v", err)
+		}
+		if string(body) != `{"price":1}` {
+			t.Fatalf("unexpected body: %s", body)
+		}
+	}
+	if got := atomic.LoadInt32(hits); got != 1 {
+		t.Fatalf("expected 1 upstream hit, got %d", got)
+	}
+}
+
+func TestFetchJSON_RefetchesAfterTTL(t *testing.T) {
+	withSharedCache(t, 20*time.Millisecond)
+	srv, hits := countingServer(t, `{"price":1}`)
+	r := newTestReader(t, srv.URL)
+
+	if _, err := r.FetchJSON(context.Background()); err != nil {
+		t.Fatalf("FetchJSON: %v", err)
+	}
+	time.Sleep(40 * time.Millisecond)
+	if _, err := r.FetchJSON(context.Background()); err != nil {
+		t.Fatalf("FetchJSON: %v", err)
+	}
+	if got := atomic.LoadInt32(hits); got != 2 {
+		t.Fatalf("expected 2 upstream hits after TTL expiry, got %d", got)
+	}
+}
+
+func TestFetchJSON_DisabledPassesThrough(t *testing.T) {
+	withSharedCache(t, 0)
+	srv, hits := countingServer(t, `{"price":1}`)
+	r := newTestReader(t, srv.URL)
+
+	for i := 0; i < 3; i++ {
+		if _, err := r.FetchJSON(context.Background()); err != nil {
+			t.Fatalf("FetchJSON: %v", err)
+		}
+	}
+	if got := atomic.LoadInt32(hits); got != 3 {
+		t.Fatalf("expected 3 upstream hits when cache disabled, got %d", got)
+	}
+}
+
+func TestFetchJSON_CoalescesConcurrentRequests(t *testing.T) {
+	withSharedCache(t, time.Minute)
+	var hits int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		atomic.AddInt32(&hits, 1)
+		time.Sleep(80 * time.Millisecond) // hold the connection so callers overlap
+		_, _ = w.Write([]byte(`{"price":1}`))
+	}))
+	t.Cleanup(srv.Close)
+	r := newTestReader(t, srv.URL)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := r.FetchJSON(context.Background()); err != nil {
+				t.Errorf("FetchJSON: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+	if got := atomic.LoadInt32(&hits); got != 1 {
+		t.Fatalf("expected concurrent requests to coalesce into 1 upstream hit, got %d", got)
+	}
+}
+
+func TestCacheKey_DistinctByURLQueryAndHeaders(t *testing.T) {
+	mk := func(url, query string, headers map[string]string) string {
+		r, err := NewReader(url, http.MethodGet, query, headers, nil, 5000, nil)
+		if err != nil {
+			t.Fatalf("NewReader: %v", err)
+		}
+		return r.cacheKey()
+	}
+	base := mk("http://x/a", "", nil)
+	cases := map[string]string{
+		"different url":    mk("http://x/b", "", nil),
+		"different query":  mk("http://x/a", "{q}", nil),
+		"different header": mk("http://x/a", "", map[string]string{"k": "v"}),
+	}
+	for name, key := range cases {
+		if key == base {
+			t.Errorf("cacheKey should differ for %s but matched base", name)
+		}
+	}
+	// Same inputs must produce the same key.
+	if mk("http://x/a", "", nil) != base {
+		t.Error("cacheKey not stable for identical inputs")
+	}
+}

custom_query/rpc/rpc_reader/reader.go

@@ -70,7 +70,9 @@ func cloneStringMap(m map[string]string) map[string]string {
 	return out
 }
 
-func (r *Reader) FetchJSON(ctx context.Context) ([]byte, error) {
+// fetchWithRetry performs the actual HTTP fetch (with retries). FetchJSON
+// (see cache.go) wraps this with the optional in-memory response cache.
+func (r *Reader) fetchWithRetry(ctx context.Context) ([]byte, error) {
 	startTime := time.Now()
 	defer func() {
 		metrics.RPCCallDuration.Observe(time.Since(startTime).Seconds())

go.mod

@@ -233,7 +233,7 @@ require (
 	golang.org/x/crypto v0.46.0
 	golang.org/x/net v0.48.0 // indirect
 	golang.org/x/oauth2 v0.26.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sync v0.19.0
 	golang.org/x/sys v0.39.0 // indirect
 	golang.org/x/term v0.38.0 // indirect
 	golang.org/x/time v0.9.0 // indirect