feat(beacon): add streaming raw API responses

.github/workflows/test.yaml

@@ -22,6 +22,14 @@ jobs:
       - name: run tests
         run: go test -json ./... > test.json
 
+      - name: run bounded-memory acceptance tests
+        run: |
+          go test -tags acceptance ./pkg/beacon/api -run Acceptance -count=1 -timeout=20m
+          go test -tags "acceptance acceptance_http2" ./pkg/beacon/api -run TestAcceptanceConcurrentHTTP2BeaconStateStreams -count=1 -timeout=20m
+
+      - name: run race tests
+        run: go test -race ./...
+
       - name: Annotate tests
         if: always()
         uses: guyarb/golang-test-annotations@96fc379b171c49932041d6c789e73331a7bdeec1 # v0.9.0

.golangci.yml

@@ -1,5 +1,9 @@
 version: "2"
 run:
+  build-tags:
+    - acceptance
+    - acceptance_full_buffered
+    - acceptance_http2
   issues-exit-code: 1
 linters:
   default: none

README.md

@@ -27,6 +27,45 @@ go get github.com/ethpandaops/beacon
 
 ## Usage
 
+### Streaming raw responses
+
+The `OpenRaw*` methods return after the node accepts the request and sends response headers. The caller must close the response and treat any later read error as a failed transfer.
+
+```go
+func archiveState(ctx context.Context, node beacon.Node, dst io.Writer) error {
+	response, err := node.OpenRawBeaconState(ctx, "finalized", "application/octet-stream")
+	if err != nil {
+		return err
+	}
+	defer response.Close()
+
+	mediaType, _, err := mime.ParseMediaType(response.ContentType)
+	if err != nil || mediaType != "application/octet-stream" {
+		return fmt.Errorf("unexpected content type %q", response.ContentType)
+	}
+
+	written, err := io.Copy(dst, response)
+	if err != nil {
+		return err
+	}
+	if response.ContentLength >= 0 && written != response.ContentLength {
+		return fmt.Errorf("copied %d bytes, expected %d", written, response.ContentLength)
+	}
+
+	return nil
+}
+```
+
+The request context and configured raw API client timeout both govern body reads. The raw client settings apply to both `FetchRaw*` and `OpenRaw*` calls; set `SetAPIClientTimeout(0)` only when every call has a suitable context deadline. By default the raw client permits 32 active response bodies and 32 connections per host; `MaxConcurrentRequests` also caps multiplexed HTTP/2 streams.
+
+Go may transparently decompress gzip responses, reported by `RawResponse.Uncompressed`. Call `Options.SetAPIClientAcceptEncoding("gzip")` when the caller needs the encoded wire bytes; the returned headers then retain `Content-Encoding`. A `ContentLength` of `-1` means the caller cannot verify an expected byte count from that field. Slow readers keep upstream resources open, so applications should bound concurrency and finish downstream writes before committing their output.
+
+Maintainers can run the streaming memory gates with `go test -tags acceptance ./pkg/beacon/api -run Acceptance -count=1`. The full 50-response buffered comparison requires at least 10 GiB of available memory and is intentionally manual:
+
+```bash
+go test -tags "acceptance acceptance_full_buffered" ./pkg/beacon/api -run TestAcceptanceFullBufferedBaseline -count=1 -timeout=20m
+```
+
 ### Simple example
 
 ```go

pkg/beacon/api/api.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"net/url"
 
 	"github.com/ethpandaops/beacon/pkg/beacon/api/types"
 	"github.com/sirupsen/logrus"
@@ -24,27 +23,52 @@ type ConsensusClient interface {
 	NodePeer(ctx context.Context, peerID string) (types.Peer, error)
 	NodePeers(ctx context.Context) (types.Peers, error)
 	NodePeerCount(ctx context.Context) (types.PeerCount, error)
-	RawBlock(ctx context.Context, stateID string, contentType string) ([]byte, error)
+	RawBlock(ctx context.Context, blockID string, contentType string) ([]byte, error)
 	RawExecutionPayloadEnvelope(ctx context.Context, blockID string, contentType string) ([]byte, error)
 	RawDebugBeaconState(ctx context.Context, stateID string, contentType string) ([]byte, error)
+	// OpenRawBlock opens a successful raw block response without consuming its body.
+	OpenRawBlock(ctx context.Context, blockID string, contentType string) (*RawResponse, error)
+	// OpenRawExecutionPayloadEnvelope opens a successful raw envelope response without consuming its body.
+	OpenRawExecutionPayloadEnvelope(ctx context.Context, blockID string, contentType string) (*RawResponse, error)
+	// OpenRawDebugBeaconState opens a successful raw beacon state response without consuming its body.
+	OpenRawDebugBeaconState(ctx context.Context, stateID string, contentType string) (*RawResponse, error)
 	DepositSnapshot(ctx context.Context) (*types.DepositSnapshot, error)
 	NodeIdentity(ctx context.Context) (*types.Identity, error)
 }
 
 type consensusClient struct {
-	url     string
-	log     logrus.FieldLogger
-	client  http.Client
-	headers map[string]string
+	url                 string
+	log                 logrus.FieldLogger
+	client              *http.Client
+	rawClient           *http.Client
+	headers             map[string]string
+	rawResponseObserver RawResponseObserver
 }
 
-// NewConsensusClient creates a new ConsensusClient.
-func NewConsensusClient(ctx context.Context, log logrus.FieldLogger, url string, client http.Client, headers map[string]string) ConsensusClient {
+// NewConsensusClient creates a ConsensusClient. Both HTTP clients must be non-nil.
+func NewConsensusClient(
+	log logrus.FieldLogger,
+	url string,
+	client *http.Client,
+	rawClient *http.Client,
+	headers map[string]string,
+	rawResponseObserver RawResponseObserver,
+) ConsensusClient {
+	if client == nil {
+		panic("api: nil HTTP client")
+	}
+
+	if rawClient == nil {
+		panic("api: nil raw HTTP client")
+	}
+
 	return &consensusClient{
-		url:     url,
-		log:     log,
-		client:  client,
-		headers: headers,
+		url:                 url,
+		log:                 log,
+		client:              client,
+		rawClient:           rawClient,
+		headers:             headers,
+		rawResponseObserver: rawResponseObserver,
 	}
 }
 
@@ -130,43 +154,19 @@ func (c *consensusClient) get(ctx context.Context, path string) (json.RawMessage
 }
 
 func (c *consensusClient) getRaw(ctx context.Context, path string, contentType string) ([]byte, error) {
-	if contentType == "" {
-		contentType = "application/json"
-	}
-
-	u, err := url.Parse(c.url + path)
+	rsp, err := c.doRaw(ctx, path, contentType)
 	if err != nil {
 		return nil, err
 	}
 
-	req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set headers from c.headers
-	for k, v := range c.headers {
-		req.Header.Set(k, v)
-	}
-
-	req.Header.Set("Accept", contentType)
+	defer rsp.Body.Close()
 
-	rsp, err := c.client.Do(req)
+	data, err := io.ReadAll(rsp.Body)
 	if err != nil {
 		return nil, err
 	}
 
-	defer rsp.Body.Close()
-
-	if rsp.StatusCode != http.StatusOK {
-		if rsp.StatusCode == http.StatusNotFound {
-			return nil, fmt.Errorf("status code: %d: %w", rsp.StatusCode, ErrNotFound)
-		}
-
-		return nil, fmt.Errorf("status code: %d", rsp.StatusCode)
-	}
-
-	return io.ReadAll(rsp.Body)
+	return data, nil
 }
 
 // NodePeers returns the list of peers connected to the node.
@@ -225,8 +225,8 @@ func (c *consensusClient) RawDebugBeaconState(ctx context.Context, stateID strin
 }
 
 // RawBlock returns the block in the requested format.
-func (c *consensusClient) RawBlock(ctx context.Context, stateID string, contentType string) ([]byte, error) {
-	data, err := c.getRaw(ctx, fmt.Sprintf("/eth/v2/beacon/blocks/%s", stateID), contentType)
+func (c *consensusClient) RawBlock(ctx context.Context, blockID string, contentType string) ([]byte, error) {
+	data, err := c.getRaw(ctx, fmt.Sprintf("/eth/v2/beacon/blocks/%s", blockID), contentType)
 	if err != nil {
 		return nil, err
 	}