Validator set streaming

README.md

@@ -136,6 +136,7 @@ redis-cli DEL boost-relay/sepolia:validators-registration boost-relay/sepolia:va
 * `DB_DONT_APPLY_SCHEMA` - disable applying DB schema on startup (useful for connecting data API to read-only replica)
 * `DB_TABLE_PREFIX` - prefix to use for db tables (default uses `dev`)
 * `GETPAYLOAD_RETRY_TIMEOUT_MS` - getPayload retry getting a payload if first try failed (default: `100`)
+* `GETVALIDATORS_TIMEOUT_SEC` - timeout for the beacon node getStateValidators request used to refresh known validators (default: `60`)
 * `MEMCACHED_URIS` - optional comma separated list of memcached endpoints, typically used as secondary storage alongside Redis
 * `MEMCACHED_EXPIRY_SECONDS` - item expiry timeout when using memcache (default: `45`)
 * `MEMCACHED_CLIENT_TIMEOUT_MS` - client timeout in milliseconds (default: `250`)
@@ -155,6 +156,7 @@ redis-cli DEL boost-relay/sepolia:validators-registration boost-relay/sepolia:va
 * `ENABLE_IGNORABLE_VALIDATION_ERRORS` - enable ignorable validation errors
 * `USE_V1_PUBLISH_BLOCK_ENDPOINT` - uses the v1 publish block endpoint on the beacon node
 * `USE_SSZ_ENCODING_PUBLISH_BLOCK` - uses the SSZ encoding for the publish block endpoint
+* `USE_STREAM_DECODING_GET_VALIDATORS` - decodes the getStateValidators response as a stream instead of buffering it, which substantially lowers peak memory during the known-validator refresh
 
 #### Development Environment Variables
 

beaconclient/prod_beacon_instance.go

@@ -163,7 +163,9 @@ type ValidatorResponseValidatorData struct {
 func (c *ProdBeaconInstance) GetStateValidators(stateID string) (*GetStateValidatorsResponse, error) {
 	uri := fmt.Sprintf("%s/eth/v1/beacon/states/%s/validators?status=active,pending", c.beaconURI, stateID)
 	vd := new(GetStateValidatorsResponse)
-	_, err := fetchBeacon(http.MethodGet, uri, nil, vd, nil, http.Header{}, false)
+	// Passing a nil client here would fall back to http.DefaultClient, which has no
+	// timeout - see stateValidatorsClient.
+	_, err := fetchBeacon(http.MethodGet, uri, nil, vd, stateValidatorsClient, http.Header{}, false)
 	return vd, err
 }
 

beaconclient/state_validators_stream.go

@@ -0,0 +1,156 @@
+package beaconclient
+
+// Streaming alternative to GetStateValidators, used by the known-validator refresh.
+//
+// The response is hundreds of megabytes of JSON on mainnet, and io.ReadAll holds all
+// of it in memory for as long as json.Unmarshal runs. Decoding one validator at a
+// time means that body never exists.
+//
+// Everything the streaming path needs is in this file and nothing outside it changes,
+// so the whole thing can be reverted by deleting it. StreamStateValidators is
+// deliberately not on IMultiBeaconClient/IBeaconInstance - the datastore reaches it by
+// type assertion, so no other implementation (including the mocks) has to know it exists.
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/flashbots/go-utils/cli"
+	"github.com/flashbots/mev-boost-relay/common"
+	"github.com/goccy/go-json"
+)
+
+// ErrStreamStateValidatorsUnsupported is returned when no configured beacon instance
+// can stream, so the caller can fall back to GetStateValidators.
+var ErrStreamStateValidatorsUnsupported = errors.New("no beacon instance supports streaming state validators")
+
+// stateValidatorsClient bounds the whole validators request, response body included.
+// http.DefaultClient has no timeout, so a beacon node that accepts the connection and
+// then stops sending blocks the refresh indefinitely - and because the caller holds
+// knownValidatorsIsUpdating for the duration, no later refresh can start either. The
+// timeout has to stay well below the refresh interval so the next one runs cleanly.
+var stateValidatorsClient = &http.Client{
+	Timeout: time.Duration(cli.GetEnvInt("GETVALIDATORS_TIMEOUT_SEC", 60)) * time.Second,
+}
+
+// ValidatorCollector accumulates validators as they are decoded, so the validator set
+// is never held in its serialized form.
+type ValidatorCollector interface {
+	// Reset discards everything collected so far. It is called once before the first
+	// validator of an attempt, so a beacon node that fails part-way through leaves no
+	// partial state behind for the next one.
+	Reset()
+	Add(index uint64, pubkey string)
+}
+
+type stateValidatorStreamer interface {
+	StreamStateValidators(stateID string, collector ValidatorCollector) error
+}
+
+// StreamStateValidators streams all known validators to the collector, querying the
+// beacon nodes in least-used order because it is a heavy call on the CL.
+func (c *MultiBeaconClient) StreamStateValidators(stateID string, collector ValidatorCollector) error {
+	supported := false
+	for i, client := range c.beaconInstancesByLeastUsed() {
+		streamer, ok := client.(stateValidatorStreamer)
+		if !ok {
+			continue
+		}
+		supported = true
+
+		log := c.log.WithField("uri", client.GetURI())
+		log.Debug("fetching validators")
+		if err := streamer.StreamStateValidators(stateID, collector); err != nil {
+			log.WithError(err).Error("failed to fetch validators")
+			continue
+		}
+
+		c.bestBeaconIndex.Store(int64(i))
+		return nil
+	}
+
+	if !supported {
+		return ErrStreamStateValidatorsUnsupported
+	}
+	return ErrBeaconNodesUnavailable
+}
+
+// StreamStateValidators loads all active and pending validators, handing each to the
+// collector as it is decoded.
+// https://ethereum.github.io/beacon-APIs/#/Beacon/getStateValidators
+func (c *ProdBeaconInstance) StreamStateValidators(stateID string, collector ValidatorCollector) error {
+	uri := fmt.Sprintf("%s/eth/v1/beacon/states/%s/validators?status=active,pending", c.beaconURI, stateID)
+
+	req, err := http.NewRequest(http.MethodGet, uri, nil)
+	if err != nil {
+		return fmt.Errorf("invalid request for %s: %w", uri, err)
+	}
+	req.Header.Set("Accept", common.ApplicationJSON)
+
+	resp, err := stateValidatorsClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("client refused for %s: %w", uri, err)
+	}
+	defer resp.Body.Close() //nolint:errcheck
+
+	if resp.StatusCode >= http.StatusMultipleChoices {
+		// Bounded, because we do not want to buffer an unbounded error body either.
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 8*1024))
+		return fmt.Errorf("%w: %s: %s", ErrHTTPErrorResponse, uri, body)
+	}
+
+	return decodeStateValidators(resp.Body, collector)
+}
+
+// decodeStateValidators hands every entry of the response's "data" array to the
+// collector, decoding one at a time so the body is never held in full.
+func decodeStateValidators(body io.Reader, collector ValidatorCollector) error {
+	dec := json.NewDecoder(body)
+
+	// Walk to the "data" array. The only other fields the beacon API puts here are
+	// execution_optimistic and finalized, both booleans, so matching on the token
+	// value alone cannot collide with a field value.
+	for {
+		token, err := dec.Token()
+		if err != nil {
+			return fmt.Errorf("validators response has no data field: %w", err)
+		}
+		if token == "data" {
+			break
+		}
+	}
+	// Check this is really the opening bracket rather than just consuming a token:
+	// if "data" ever held a scalar, null, or an object, that value would be consumed
+	// here, dec.More() would immediately be false, and an EMPTY validator set would be
+	// returned with no error - and then installed as the authoritative map.
+	token, err := dec.Token()
+	if err != nil {
+		return fmt.Errorf("malformed validators response: %w", err)
+	}
+	if delim, ok := token.(json.Delim); !ok || delim != '[' {
+		return fmt.Errorf("validators response data is not an array, got %v", token)
+	}
+
+	// Reset only once the validators are about to arrive, so a node that fails before
+	// this point leaves the collector untouched.
+	collector.Reset()
+	for dec.More() {
+		var entry ValidatorResponseEntry
+		if err := dec.Decode(&entry); err != nil {
+			return fmt.Errorf("could not decode validator: %w", err)
+		}
+		collector.Add(entry.Index, entry.Validator.Pubkey)
+	}
+
+	// This check is load-bearing, not tidiness: dec.More() also returns false when the
+	// stream ends early, so without reading the closing bracket a connection dropped
+	// mid-array would look like a complete - but truncated - validator set, and get
+	// installed as the authoritative one.
+	if _, err := dec.Token(); err != nil {
+		return fmt.Errorf("truncated validators response: %w", err)
+	}
+	return nil
+}

beaconclient/state_validators_stream_test.go

@@ -0,0 +1,214 @@
+package beaconclient
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/flashbots/mev-boost-relay/common"
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/require"
+)
+
+type collectorForTest struct {
+	byPubkey map[string]uint64
+	resets   int
+}
+
+func newCollectorForTest() *collectorForTest {
+	return &collectorForTest{byPubkey: make(map[string]uint64)}
+}
+
+func (c *collectorForTest) Reset() {
+	c.byPubkey = make(map[string]uint64)
+	c.resets++
+}
+
+func (c *collectorForTest) Add(index uint64, pubkey string) {
+	c.byPubkey[pubkey] = index
+}
+
+func testPubkey(index int) string { return fmt.Sprintf("0x%096x", index) }
+
+// validatorsJSON renders a response with all the fields a real CL sends, including the
+// ones the relay discards.
+func validatorsJSON(numValidators int) string {
+	var b strings.Builder
+	b.WriteString(`{"execution_optimistic":false,"finalized":true,"data":[`)
+	for i := range numValidators {
+		if i > 0 {
+			b.WriteString(",")
+		}
+		fmt.Fprintf(&b, `{"index":"%d","balance":"32000000000","status":"active_ongoing",`+
+			`"validator":{"pubkey":"%s","withdrawal_credentials":"0x00e0f9b1",`+
+			`"effective_balance":"32000000000","slashed":false,`+
+			`"activation_eligibility_epoch":"1","activation_epoch":"2",`+
+			`"exit_epoch":"18446744073709551615","withdrawable_epoch":"18446744073709551615"}}`,
+			i, testPubkey(i))
+	}
+	b.WriteString("]}")
+	return b.String()
+}
+
+func TestDecodeStateValidators(t *testing.T) {
+	t.Run("decodes every entry with the right index", func(t *testing.T) {
+		collector := newCollectorForTest()
+		require.NoError(t, decodeStateValidators(strings.NewReader(validatorsJSON(3)), collector))
+
+		require.Len(t, collector.byPubkey, 3)
+		require.Equal(t, 1, collector.resets)
+		for i := range 3 {
+			index, found := collector.byPubkey[testPubkey(i)]
+			require.True(t, found, "missing validator %d", i)
+			require.Equal(t, uint64(i), index) //nolint:gosec
+		}
+	})
+
+	t.Run("handles an empty validator set", func(t *testing.T) {
+		collector := newCollectorForTest()
+		require.NoError(t, decodeStateValidators(strings.NewReader(`{"data":[]}`), collector))
+		require.Empty(t, collector.byPubkey)
+	})
+
+	// The important one: a connection that drops before the validator array closes must
+	// be an error, never a silently truncated set that then gets installed as the
+	// authoritative validator map. Swept over every possible cut point rather than a
+	// few samples, because a single accepted truncation is a missed slot.
+	t.Run("rejects every truncation before the array closes", func(t *testing.T) {
+		full := validatorsJSON(5)
+		closingBracket := strings.LastIndex(full, "]")
+
+		for cut := range closingBracket + 1 {
+			collector := newCollectorForTest()
+			err := decodeStateValidators(strings.NewReader(full[:cut]), collector)
+			require.Error(t, err, "truncation at %d of %d bytes was accepted", cut, len(full))
+		}
+	})
+
+	// Losing only the outer closing brace is harmless: the array closed, so every
+	// validator did arrive. Rejecting it would fail a refresh for no reason.
+	t.Run("accepts a response missing only the outer closing brace", func(t *testing.T) {
+		full := validatorsJSON(5)
+		collector := newCollectorForTest()
+		require.NoError(t, decodeStateValidators(strings.NewReader(full[:len(full)-1]), collector))
+		require.Len(t, collector.byPubkey, 5)
+	})
+
+	t.Run("rejects malformed responses", func(t *testing.T) {
+		for name, body := range map[string]string{
+			"empty":         ``,
+			"no data field": `{"execution_optimistic":false}`,
+			"bad entry":     `{"data":[{"index":"not-a-number"}]}`,
+		} {
+			t.Run(name, func(t *testing.T) {
+				collector := newCollectorForTest()
+				require.Error(t, decodeStateValidators(strings.NewReader(body), collector))
+			})
+		}
+	})
+
+	// If "data" ever holds something other than an array, that must be a loud error and
+	// never an error-free empty validator set - an empty set would be installed as the
+	// authoritative map and fail every getPayload. Stock Lighthouse and Prysm cannot
+	// produce these today; this guards the case where that stops being true.
+	t.Run("rejects data that is not an array", func(t *testing.T) {
+		for name, body := range map[string]string{
+			"null":             `{"execution_optimistic":false,"data":null}`,
+			"number":           `{"execution_optimistic":false,"data":123}`,
+			"string":           `{"execution_optimistic":false,"data":"nope"}`,
+			"object":           `{"execution_optimistic":false,"data":{}}`,
+			"bool":             `{"execution_optimistic":false,"data":true}`,
+			"null then fields": `{"data":null,"execution_optimistic":false}`,
+		} {
+			t.Run(name, func(t *testing.T) {
+				collector := newCollectorForTest()
+				err := decodeStateValidators(strings.NewReader(body), collector)
+				require.Error(t, err, "a non-array data field was accepted as an empty validator set")
+			})
+		}
+	})
+}
+
+func TestProdBeaconInstanceStreamStateValidators(t *testing.T) {
+	newServer := func(t *testing.T, handler http.HandlerFunc) string {
+		t.Helper()
+		r := mux.NewRouter()
+		srv := httptest.NewServer(r)
+		t.Cleanup(srv.Close)
+		r.HandleFunc("/eth/v1/beacon/states/{state_id}/validators", handler)
+		return srv.URL
+	}
+
+	t.Run("streams from the beacon node", func(t *testing.T) {
+		url := newServer(t, func(w http.ResponseWriter, req *http.Request) {
+			require.Equal(t, "active,pending", req.URL.Query().Get("status"))
+			_, err := io.WriteString(w, validatorsJSON(4))
+			require.NoError(t, err)
+		})
+
+		collector := newCollectorForTest()
+		bc := NewProdBeaconInstance(common.TestLog, url, url)
+		require.NoError(t, bc.StreamStateValidators("head", collector))
+		require.Len(t, collector.byPubkey, 4)
+	})
+
+	t.Run("surfaces beacon node errors", func(t *testing.T) {
+		url := newServer(t, func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusInternalServerError)
+			_, err := io.WriteString(w, `{"code":500,"message":"state not found"}`)
+			require.NoError(t, err)
+		})
+
+		collector := newCollectorForTest()
+		bc := NewProdBeaconInstance(common.TestLog, url, url)
+		err := bc.StreamStateValidators("head", collector)
+		require.ErrorIs(t, err, ErrHTTPErrorResponse)
+		require.ErrorContains(t, err, "state not found")
+		require.Empty(t, collector.byPubkey)
+	})
+}
+
+func TestMultiBeaconClientStreamStateValidators(t *testing.T) {
+	// The mocks do not implement the streaming path, so the multi client must say so
+	// rather than silently returning an empty validator set.
+	t.Run("reports when no instance can stream", func(t *testing.T) {
+		bc := NewMultiBeaconClient(common.TestLog, []IBeaconInstance{NewMockBeaconInstance()})
+
+		collector := newCollectorForTest()
+		require.ErrorIs(t, bc.StreamStateValidators("head", collector), ErrStreamStateValidatorsUnsupported)
+		require.Empty(t, collector.byPubkey)
+	})
+
+	t.Run("falls through to the next node, discarding the partial set", func(t *testing.T) {
+		r := mux.NewRouter()
+		srv := httptest.NewServer(r)
+		t.Cleanup(srv.Close)
+
+		// Dies after two validators, leaving the collector holding a partial set.
+		r.HandleFunc("/broken/eth/v1/beacon/states/{state_id}/validators", func(w http.ResponseWriter, _ *http.Request) {
+			partial := validatorsJSON(3)
+			_, err := io.WriteString(w, partial[:len(partial)/2])
+			require.NoError(t, err)
+		})
+		r.HandleFunc("/good/eth/v1/beacon/states/{state_id}/validators", func(w http.ResponseWriter, _ *http.Request) {
+			_, err := io.WriteString(w, validatorsJSON(1))
+			require.NoError(t, err)
+		})
+
+		// beaconInstancesByLeastUsed reverses the order, so the broken node goes first.
+		bc := NewMultiBeaconClient(common.TestLog, []IBeaconInstance{
+			NewProdBeaconInstance(common.TestLog, srv.URL+"/good", srv.URL+"/good"),
+			NewProdBeaconInstance(common.TestLog, srv.URL+"/broken", srv.URL+"/broken"),
+		})
+
+		collector := newCollectorForTest()
+		require.NoError(t, bc.StreamStateValidators("head", collector))
+
+		require.Len(t, collector.byPubkey, 1)
+		require.Contains(t, collector.byPubkey, testPubkey(0))
+		require.Equal(t, 2, collector.resets, "each attempt must reset before emitting")
+	})
+}