Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 36 additions & 2 deletions harnesses/hyperliquid-frontends/cmd/script/agg.go
Original file line number Diff line number Diff line change
Expand Up @@ -131,8 +131,16 @@ func (a *Aggregator) runJobs(ctx context.Context, jobs []fetchJob) int {
if ctx.Err() != nil {
return
}
had := a.mirror.present(j.addr, j.day)
err := a.mirror.fetch(ctx, j.addr, j.day)
if err != nil && err != errFeedAbsent {
if err == nil && had {
// Revalidation returned a new body: the CDN completed a
// day it had published short. The parse we cached is of
// the old half, so drop it.
a.forgetDay(j.addr, j.day)
log.Printf("feed %s %s: upstream republished, cached parse dropped", j.addr, j.day.Format(dayFormat))
}
if err != nil && err != errFeedAbsent && err != errFeedUnchanged {
fmu.Lock()
failures++
fmu.Unlock()
Expand Down Expand Up @@ -166,7 +174,21 @@ func (a *Aggregator) syncWindow(ctx context.Context) {
recent := d.After(to.AddDate(0, 0, -a.graceDays))
for _, addr := range a.addrs {
if a.mirror.present(addr, d) {
continue
// Not "we have it, we are done". Hyperliquid publishes a day
// short and completes it later: 3 October was still a 403 at
// 22:00 on the 4th and was whole by 01:00 on the 5th, while
// 22 September to 2 October stopped near 12:11 UTC and have
// never moved since. Holding the first copy for ever is how a
// half day becomes permanent on our side.
//
// Revalidation is conditional (If-Modified-Since from the
// file's own mtime), so an unchanged day costs a 304 and
// nothing else. Only days still inside the grace window are
// re-checked: past it the upstream has never once gone back
// and filled a day in, so it would be pure load.
if !recent {
continue
}
}
// A 403 on an old day is final (no fills). On a recent day every
// missing file is re-requested on every pass, so by the time the
Expand Down Expand Up @@ -893,3 +915,15 @@ func (a *Aggregator) publishCoverage(D, windowStart time.Time, lastFill map[stri
}
hlTruncatedDaysWindow.Set(float64(truncated))
}

// forgetDay drops one cached parse so the next read re-reads the file from
// disk. Used when a revalidation replaces a mirrored day with a longer one.
func (a *Aggregator) forgetDay(addr string, day time.Time) {
a.mu.Lock()
defer a.mu.Unlock()
// Keyed exactly as summary() keys it: the address as the caller passes
// it, which is the form a.addrs carries.
if m, ok := a.cache[addr]; ok {
delete(m, dayKey(day))
}
}
21 changes: 20 additions & 1 deletion harnesses/hyperliquid-frontends/cmd/script/feed.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,20 @@ import (
// 02:45 to 03:30 UTC). A missing file answers 403, which is also what a
// builder with no fills that day gets, so "not published yet" and "no
// fills" are told apart at cohort level (see availability.go).
const feedURLTemplate = "https://stats-data.hyperliquid.xyz/Mainnet/builder_fills/%s/%s.csv.lz4"
// feedURLTemplate is a var, not a const, so the revalidation path can be
// exercised against a local server in tests. Never reassigned in production.
var feedURLTemplate = "https://stats-data.hyperliquid.xyz/Mainnet/builder_fills/%s/%s.csv.lz4"

const dayFormat = "20060102"

var feedHTTP = &http.Client{Timeout: 120 * time.Second}

var errFeedAbsent = errors.New("feed: no file for that builder and day")

// errFeedUnchanged is the CDN answering 304 to a revalidation: the mirrored
// copy is still the newest one published. Not a failure.
var errFeedUnchanged = errors.New("feed: not modified")

// Mirror keeps a local copy of every fetched CSV under
// <root>/builder_fills/<address>/<YYYYMMDD>.csv.lz4, plus a
// <YYYYMMDD>.absent marker (containing the unix time of the check) when the
Expand Down Expand Up @@ -129,13 +135,26 @@ func (m *Mirror) fetch(ctx context.Context, addr string, day time.Time) error {
return err
}
req.Header.Set("User-Agent", "openchainbench-hl-frontends/2 (+https://openchainbench.com)")
// Revalidate rather than re-download. The mirrored file already
// carries the CDN's own Last-Modified as its mtime (stamped below),
// so it is the conditional token and no extra state is needed. The
// CDN answers 304, which is what makes it affordable to re-check
// every recent day on every pass instead of trusting the first copy
// for ever.
if st, err := os.Stat(m.filePath(addr, day)); err == nil && st.Size() > 0 {
req.Header.Set("If-Modified-Since", st.ModTime().UTC().Format(http.TimeFormat))
}
resp, err := feedHTTP.Do(req)
if err != nil {
lastErr = err
feedFetchTotal.WithLabelValues("error").Inc()
continue
}
switch {
case resp.StatusCode == http.StatusNotModified:
resp.Body.Close()
feedFetchTotal.WithLabelValues("unchanged").Inc()
return errFeedUnchanged
case resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusNotFound:
resp.Body.Close()
feedFetchTotal.WithLabelValues("absent").Inc()
Expand Down
102 changes: 102 additions & 0 deletions harnesses/hyperliquid-frontends/cmd/script/refetch_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
package main

import (
"context"
"net/http"
"net/http/httptest"
"os"
"testing"
"time"
)

// The behaviour under test is the one that let a half day become permanent:
// the mirror used to take the first copy of a day and never look again, so
// when Hyperliquid published 3 October short and completed it two days
// later, our side would have kept the short one for ever.
func TestRevalidationPicksUpACompletedDay(t *testing.T) {
short := []byte("short-day-body")
long := []byte("the-same-day-once-upstream-finished-writing-it")

body := short
modified := time.Date(2026, 10, 5, 0, 59, 25, 0, time.UTC)
var conditional, served int

srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if ims := r.Header.Get("If-Modified-Since"); ims != "" {
conditional++
if since, err := http.ParseTime(ims); err == nil && !modified.After(since) {
w.WriteHeader(http.StatusNotModified)
return
}
}
served++
w.Header().Set("Last-Modified", modified.UTC().Format(http.TimeFormat))
w.WriteHeader(http.StatusOK)
_, _ = w.Write(body)
}))
defer srv.Close()

old := feedURLTemplate
feedURLTemplate = srv.URL + "/%s/%s.csv.lz4"
defer func() { feedURLTemplate = old }()

m, err := newMirror(t.TempDir())
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
addr, day := "0xabc", time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC)

// First pass: nothing on disk, so no condition is sent and we take the
// short body the upstream has so far.
if err := m.fetch(ctx, addr, day); err != nil {
t.Fatalf("first fetch: %v", err)
}
if conditional != 0 {
t.Errorf("first fetch sent a condition with nothing on disk")
}
if got, _ := os.ReadFile(m.filePath(addr, day)); string(got) != string(short) {
t.Fatalf("first fetch stored %q", got)
}

// Second pass: unchanged upstream. This must cost a 304 and leave the
// file alone, which is what makes re-checking every recent day cheap
// enough to do on every sync.
if err := m.fetch(ctx, addr, day); err != errFeedUnchanged {
t.Fatalf("revalidation of an unchanged day = %v, want errFeedUnchanged", err)
}
if conditional != 1 || served != 1 {
t.Errorf("conditional=%d served=%d, want 1 and 1: the body was re-sent", conditional, served)
}

// Upstream completes the day.
body = long
modified = modified.Add(2 * time.Hour)
if err := m.fetch(ctx, addr, day); err != nil {
t.Fatalf("fetch after upstream republished: %v", err)
}
got, _ := os.ReadFile(m.filePath(addr, day))
if string(got) != string(long) {
t.Fatalf("after the upstream completed the day the mirror still holds %q", got)
}
}

// A day the upstream never had must still read as absent, not as unchanged:
// the two are different answers and the caller treats them differently.
func TestAbsentDayIsNotUnchanged(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
defer srv.Close()
old := feedURLTemplate
feedURLTemplate = srv.URL + "/%s/%s.csv.lz4"
defer func() { feedURLTemplate = old }()

m, err := newMirror(t.TempDir())
if err != nil {
t.Fatal(err)
}
if err := m.fetch(context.Background(), "0xabc", time.Date(2026, 10, 9, 0, 0, 0, 0, time.UTC)); err != errFeedAbsent {
t.Fatalf("403 = %v, want errFeedAbsent", err)
}
}
Loading