From 2a4fb89edcf313bd52f47c1ff06907a11ac4c0e3 Mon Sep 17 00:00:00 2001 From: Flotapponnier Date: Tue, 15 Sep 2026 01:29:54 +0200 Subject: [PATCH] perp-volume-history: sources run concurrently; dYdX honours 429 and gives up early The dYdX indexer rate-limited the VPS IP mid-backfill and the source spent an hour walking every market through the retry ladder while the four venues behind it waited. Sources now run in parallel (one goroutine each, flush + publish as each lands), 429s honour Retry-After, and the dYdX walk stops after a run of rate-limited calls at one request a second. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HJgbZCqjR4nvCfcJSzofbw --- .../perp-volume-history/cmd/script/main.go | 30 ++++++++++++------- .../cmd/script/source_venues.go | 16 +++++++++- .../perp-volume-history/cmd/script/sources.go | 27 ++++++++++++++++- 3 files changed, 60 insertions(+), 13 deletions(-) diff --git a/harnesses/perp-volume-history/cmd/script/main.go b/harnesses/perp-volume-history/cmd/script/main.go index 47e99f1e..8dd6410c 100644 --- a/harnesses/perp-volume-history/cmd/script/main.go +++ b/harnesses/perp-volume-history/cmd/script/main.go @@ -24,6 +24,7 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" ) @@ -105,6 +106,12 @@ func runLoop(ctx context.Context, store *Store, sources []Source, backfillDays, today := utcDay(time.Now()) lastClosed := today.AddDate(0, 0, -1) + // Every source runs in its own goroutine: they are independent + // HTTP fan-outs and one slow or rate-limited venue (dYdX on + // 2026-09-14) must not hold the others back. The store and the + // gauges are mutex-safe; each venue flushes and publishes as soon + // as it lands. + var wg sync.WaitGroup for _, src := range sources { if ctx.Err() != nil { return @@ -122,18 +129,19 @@ func runLoop(ctx context.Context, store *Store, sources []Source, backfillDays, from = horizon backfill = true } - if runSource(ctx, store, src, from, lastClosed) && backfill { - store.markBackfilled(src.Slug(), horizon) - } - // Flush and publish after every source: the first sweep runs - // for the better part of an hour, a restart must not redo - // finished venues, and the bench should light up venue by - // venue rather than all at once at the end. - if err := store.flush(); err != nil { - fmt.Fprintf(os.Stderr, "store flush: %v\n", err) - } - publish(store, sources) + wg.Add(1) + go func(src Source, from, horizon time.Time, backfill bool) { + defer wg.Done() + if runSource(ctx, store, src, from, lastClosed) && backfill { + store.markBackfilled(src.Slug(), horizon) + } + if err := store.flush(); err != nil { + fmt.Fprintf(os.Stderr, "store flush: %v\n", err) + } + publish(store, sources) + }(src, from, horizon, backfill) } + wg.Wait() publish(store, sources) if err := store.flush(); err != nil { diff --git a/harnesses/perp-volume-history/cmd/script/source_venues.go b/harnesses/perp-volume-history/cmd/script/source_venues.go index cb7b178e..dcb185c3 100644 --- a/harnesses/perp-volume-history/cmd/script/source_venues.go +++ b/harnesses/perp-volume-history/cmd/script/source_venues.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "net/url" "strconv" @@ -232,6 +233,11 @@ func (s *aevoSource) Daily(ctx context.Context, from, to time.Time) (map[time.Ti // at run time; the candles are the same fills bucketed on UTC days, // which is what a backfill needs. Candles are paged 100 per call from // the most recent, so a 400-day backfill is four pages per market. +// +// The indexer rate-limits per IP and, once tripped, answers 429 for a +// while: the throttle stays at one request a second and the sweep gives +// up after a run of rate-limited calls rather than retrying every +// market through the backoff ladder (that took an hour on 2026-09-14). type dydxSource struct{ meta venueMeta } func (s *dydxSource) Slug() string { return s.meta.slug } @@ -252,11 +258,15 @@ func (s *dydxSource) Daily(ctx context.Context, from, to time.Time) (map[time.Ti return nil, err } out := map[time.Time]float64{} - th := newThrottle(300) + th := newThrottle(60) failed := 0 + limited := 0 for ticker := range markets.Markets { toISO := to.AddDate(0, 0, 1).Format(time.RFC3339) for page := 0; page < 8; page++ { + if limited >= 6 { + return nil, fmt.Errorf("indexer rate limited, giving up this sweep (%d markets done)", len(markets.Markets)-failed) + } if err := th.wait(ctx); err != nil { return out, err } @@ -270,8 +280,12 @@ func (s *dydxSource) Daily(ctx context.Context, from, to time.Time) (map[time.Ti url.PathEscape(ticker), url.QueryEscape(toISO)) if err := getJSON(ctx, u, &resp); err != nil { failed++ + if errors.Is(err, ErrRateLimited) { + limited++ + } break } + limited = 0 if len(resp.Candles) == 0 { break } diff --git a/harnesses/perp-volume-history/cmd/script/sources.go b/harnesses/perp-volume-history/cmd/script/sources.go index 919b9dbe..fbe620ea 100644 --- a/harnesses/perp-volume-history/cmd/script/sources.go +++ b/harnesses/perp-volume-history/cmd/script/sources.go @@ -4,9 +4,11 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" + "strconv" "strings" "time" ) @@ -107,6 +109,9 @@ func buildSources(llamaKey string) []Source { var httpClient = &http.Client{Timeout: 45 * time.Second} +// ErrRateLimited is returned after a 429 survived the retry ladder. +var ErrRateLimited = errors.New("rate_limited") + const userAgent = "OpenChainBench-PerpVolumeHistory/1.0 contact@openchainbench.com" func getJSON(ctx context.Context, url string, out any) error { @@ -162,7 +167,27 @@ func doJSON(ctx context.Context, method, url string, body []byte, headers map[st } raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) resp.Body.Close() - if resp.StatusCode == 429 || resp.StatusCode >= 500 { + if resp.StatusCode == 429 { + // Honour Retry-After (capped) so a rate-limited host is not + // hammered through the backoff ladder; ErrRateLimited lets + // per-market sources stop the sweep early instead of burning + // an hour on retries. + lastErr = ErrRateLimited + wait := 5 * time.Second + if ra, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && ra > 0 { + wait = time.Duration(ra) * time.Second + } + if wait > 30*time.Second { + wait = 30 * time.Second + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(wait): + } + continue + } + if resp.StatusCode >= 500 { lastErr = fmt.Errorf("status_%d", resp.StatusCode) continue }