perp-volume-history: concurrent sources, dYdX rate-limit handling

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 {

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
 			}

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
 		}