From b666f7af04426cbd839f9cb9a3a1e9ec082d7c62 Mon Sep 17 00:00:00 2001 From: syntrust Date: Sat, 10 Oct 2026 17:36:07 +0800 Subject: [PATCH 1/4] archiver: resolve blobs from the beacon block's execution block hash - decode /eth/v2/beacon/blocks/{id} per consensus version: deneb, electra and fulu through the execution payload, gloas through the signed execution payload bid. Both paths yield the execution block hash and the KZG commitments in block order; the execution block number is dropped because a gloas bid does not carry one. - find PutBlob events with an eth_getLogs blockHash query instead of a fromBlock/toBlock range. - classify failures explicitly: beacon 404 and a blob missing from local storage are 404, upstream beacon errors and undecodable block versions are 502, and an execution block L1 has not seen yet is a retryable 503. --- ethstorage/archiver/api.go | 340 +++++++++++++++---- ethstorage/archiver/api_e2e_test.go | 455 ++++++++++++++++++++++++++ ethstorage/archiver/api_parse_test.go | 307 +++++++++++++++++ ethstorage/archiver/service.go | 4 +- ethstorage/archiver/utils.go | 32 ++ ethstorage/eth/beacon_client.go | 15 +- ethstorage/eth/polling_client.go | 22 ++ ethstorage/eth/polling_client_test.go | 221 +++++++++++++ 8 files changed, 1327 insertions(+), 69 deletions(-) create mode 100644 ethstorage/archiver/api_e2e_test.go create mode 100644 ethstorage/archiver/api_parse_test.go create mode 100644 ethstorage/eth/polling_client_test.go diff --git a/ethstorage/archiver/api.go b/ethstorage/archiver/api.go index 5f4f9827..2884d4ce 100644 --- a/ethstorage/archiver/api.go +++ b/ethstorage/archiver/api.go @@ -4,14 +4,21 @@ package archiver import ( + "context" "crypto/sha256" "encoding/json" + "errors" "fmt" + "io" "math/big" "net/http" "strconv" + "strings" + "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto/kzg4844" "github.com/ethereum/go-ethereum/log" "github.com/ethstorage/go-ethstorage/ethstorage" @@ -20,6 +27,59 @@ import ( gkzg "github.com/protolambda/go-kzg/eth" ) +const ( + // beaconBlockFetchTimeout bounds a single upstream beacon node request. + beaconBlockFetchTimeout = 30 * time.Second + // maxBeaconBlockResponseSize caps how much of a beacon block response we + // read into memory (a full mainnet block is well under this). + maxBeaconBlockResponseSize = 64 << 20 + // beaconBlockVersionDeneb .. beaconBlockVersionGloas are the consensus + // versions whose blocks the archiver knows how to decode. + beaconBlockVersionDeneb = "deneb" + beaconBlockVersionElectra = "electra" + beaconBlockVersionFulu = "fulu" + beaconBlockVersionGloas = "gloas" +) + +// beaconBlock is the execution layer view of a beacon block: the execution +// block hash and the KZG commitments of its blobs, in the order they appear +// in the block. Execution block number is deliberately not part of it: Gloas +// blocks carry a signed execution payload bid instead of an execution payload, +// and the bid has no block number (see consensus-specs gloas/beacon-chain.md). +type beaconBlock struct { + version string + slot uint64 + executionHash common.Hash + kzgCommitsAll []string +} + +// beaconBlockResponse mirrors the /eth/v2/beacon/blocks/{id} envelope for every +// supported consensus version. Fields are pointers so that a missing field can +// be told apart from an empty one. +type beaconBlockResponse struct { + Version *string `json:"version"` + Data *struct { + Message *struct { + Slot *string `json:"slot"` + Body *struct { + // Deneb, Electra and Fulu + ExecutionPayload *struct { + BlockHash *string `json:"block_hash"` + } `json:"execution_payload"` + BlobKzgCommitments *[]string `json:"blob_kzg_commitments"` + // Gloas (EIP-7732): execution payload and blob commitments moved + // into the signed bid + SignedExecutionPayloadBid *struct { + Message *struct { + BlockHash *string `json:"block_hash"` + BlobKzgCommitments *[]string `json:"blob_kzg_commitments"` + } `json:"message"` + } `json:"signed_execution_payload_bid"` + } `json:"body"` + } `json:"message"` + } `json:"data"` +} + type API struct { beaconClient *eth.BeaconClient l1Source *eth.PollingClient @@ -39,23 +99,15 @@ func NewAPI(storageMgr *ethstorage.StorageManager, beaconClient *eth.BeaconClien // From the getBlobs spec: https://ethereum.github.io/beacon-APIs/beacon-node-oapi.yaml // The returned blobs are ordered based on their kzg commitments in the block. // So EthStorage needs to follow the order of kzg commitments in the block as well. -func (a *API) queryBlobs(id string, hashes []common.Hash) (*blobs.BeaconBlobs, *httpError) { - // /eth/v2/beacon/blocks - queryUrl, err := a.beaconClient.QueryUrlForV2BeaconBlock(id) - if err != nil { - a.lg.Error("Invalid beaconID", "beaconID", id, "err", err) - return nil, errUnknownBlock - } - elBlock, kzgCommitsAll, hErr := a.queryElBlockNumberAndKzg(queryUrl) +func (a *API) queryBlobs(ctx context.Context, id string, hashes []common.Hash) (*blobs.BeaconBlobs, *httpError) { + block, hErr := a.queryBeaconBlock(ctx, id) if hErr != nil { - a.lg.Error("Failed to get execution block number", "beaconID", id, "err", err) return nil, hErr } - a.lg.Info("BeaconID to execution block number", "beaconID", id, "elBlock", elBlock) // hashToIndex is used to determine correct blob index hashToIndex := make(map[common.Hash]int) - for _, c := range kzgCommitsAll { + for _, c := range block.kzgCommitsAll { var commit kzg4844.Commitment copy(commit[:], common.FromHex(c)) vh := common.Hash(kzg4844.CalcBlobHashV1(sha256.New(), &commit)) @@ -64,12 +116,10 @@ func (a *API) queryBlobs(id string, hashes []common.Hash) (*blobs.BeaconBlobs, * } } - // get event logs on the block - blockBN := new(big.Int).SetUint64(elBlock) - events, err := a.l1Source.FilterLogsByBlockRange(blockBN, blockBN, eth.PutBlobEvent) - if err != nil { - a.lg.Error("Failed to get events", "err", err) - return nil, errServerError + // get event logs on the execution block + events, hErr := a.queryPutBlobEvents(ctx, id, block) + if hErr != nil { + return nil, hErr } blobsData := make([]string, len(hashToIndex)) @@ -80,14 +130,9 @@ func (a *API) queryBlobs(id string, hashes []common.Hash) (*blobs.BeaconBlobs, * if index, ok := hashToIndex[blobHash]; ok { kvIndex := big.NewInt(0).SetBytes(event.Topics[1][:]).Uint64() a.lg.Info("Blobhash matched", "blobhash", blobHash, "index", index, "kvIndex", kvIndex) - blobData, found, err := a.storageMgr.TryRead(kvIndex, int(a.storageMgr.MaxKvSize()), blobHash) - if err != nil { - a.lg.Error("Failed to read blob", "err", err) - return nil, errServerError - } - if !found { - a.lg.Info("Blob not found by storage manager", "kvIndex", kvIndex, "blobHash", blobHash) - return nil, errBlobNotInES + blobData, hErr := a.readBlob(kvIndex, blobHash) + if hErr != nil { + return nil, hErr } blobsData[index] = "0x" + common.Bytes2Hex(blobData) } @@ -99,24 +144,17 @@ func (a *API) queryBlobs(id string, hashes []common.Hash) (*blobs.BeaconBlobs, * filteredBlobs = append(filteredBlobs, data) } } - a.lg.Info("Query blobs done", "blobs", len(filteredBlobs)) + a.lg.Info("Query blobs done", "beaconID", id, "blobs", len(filteredBlobs)) return &blobs.BeaconBlobs{Data: filteredBlobs}, nil } // Deprecated since Fusaka -func (a *API) queryBlobSidecars(id string, indices []uint64) (*BlobSidecars, *httpError) { - // query EL block - queryUrl, err := a.beaconClient.QueryUrlForV2BeaconBlock(id) - if err != nil { - a.lg.Error("Invalid beaconID", "beaconID", id, "err", err) - return nil, errUnknownBlock - } - elBlock, kzgCommitsAll, hErr := a.queryElBlockNumberAndKzg(queryUrl) +func (a *API) queryBlobSidecars(ctx context.Context, id string, indices []uint64) (*BlobSidecars, *httpError) { + block, hErr := a.queryBeaconBlock(ctx, id) if hErr != nil { - a.lg.Error("Failed to get execution block number", "beaconID", id, "err", err) return nil, hErr } - a.lg.Info("BeaconID to execution block number", "beaconID", id, "elBlock", elBlock) + kzgCommitsAll := block.kzgCommitsAll blobsInBeacon := uint64(len(kzgCommitsAll)) for _, index := range indices { @@ -135,12 +173,10 @@ func (a *API) queryBlobSidecars(id string, indices []uint64) (*BlobSidecars, *ht } } - // get event logs on the block - blockBN := new(big.Int).SetUint64(elBlock) - events, err := a.l1Source.FilterLogsByBlockRange(blockBN, blockBN, eth.PutBlobEvent) - if err != nil { - a.lg.Error("Failed to get events", "err", err) - return nil, errServerError + // get event logs on the execution block + events, hErr := a.queryPutBlobEvents(ctx, id, block) + if hErr != nil { + return nil, hErr } var res BlobSidecars for i, event := range events { @@ -160,45 +196,208 @@ func (a *API) queryBlobSidecars(id string, indices []uint64) (*BlobSidecars, *ht res.Data = append(res.Data, sidecar) } } - a.lg.Info("Query blob sidecars done", "blobs", len(res.Data)) + a.lg.Info("Query blob sidecars done", "beaconID", id, "blobs", len(res.Data)) return &res, nil } -func (a *API) queryElBlockNumberAndKzg(queryUrl string) (uint64, []string, *httpError) { - resp, err := http.Get(queryUrl) +// queryBeaconBlock fetches a beacon block and reduces it to the execution block +// hash plus its ordered KZG commitments, whatever the consensus version is. +func (a *API) queryBeaconBlock(ctx context.Context, id string) (*beaconBlock, *httpError) { + // /eth/v2/beacon/blocks + queryUrl, err := a.beaconClient.QueryUrlForV2BeaconBlock(id) if err != nil { - return 0, nil, errServerError + a.lg.Error("Invalid beaconID", "beaconID", id, "err", err) + return nil, errUnknownBlock + } + block, hErr := fetchBeaconBlock(ctx, a.beaconClient.HTTPClient(), queryUrl) + if hErr != nil { + a.lg.Error("Failed to get beacon block", "beaconID", id, "err", hErr) + return nil, hErr + } + a.lg.Info("BeaconID to execution block", "beaconID", id, "version", block.version, + "slot", block.slot, "elBlockHash", block.executionHash, "blobsInBlock", len(block.kzgCommitsAll)) + return block, nil +} + +// queryPutBlobEvents returns the PutBlob events of the execution block, ordered +// as the EL reports them. Querying by block hash (EIP-234) rather than by block +// number keeps the events aligned with the commitments of the beacon block even +// during reorgs. +func (a *API) queryPutBlobEvents(ctx context.Context, id string, block *beaconBlock) ([]types.Log, *httpError) { + events, err := a.l1Source.FilterLogsByBlockHash(ctx, block.executionHash, eth.PutBlobEvent) + if err != nil { + if isUnknownBlockErr(err) { + // the execution block the bid points to is not available on the EL yet + a.lg.Warn("Execution block not available on L1", "beaconID", id, + "version", block.version, "elBlockHash", block.executionHash, "err", err) + return nil, errExecutionBlockNotAvailable + } + a.lg.Error("Failed to get events", "beaconID", id, "elBlockHash", block.executionHash, "err", err) + return nil, errServerError + } + return events, nil +} + +// fetchBeaconBlock requests a /eth/v2/beacon/blocks/{id} response and decodes it +// into a beaconBlock. Error classification: +// - 404 from the beacon node: the block does not exist +// - transport errors, timeouts, >=500, invalid JSON: upstream error (5xx) +// - unknown version, or a known version missing required fields: upstream +// error (5xx), never a silent "block not found" +func fetchBeaconBlock(ctx context.Context, client *http.Client, queryUrl string) (*beaconBlock, *httpError) { + ctx, cancel := context.WithTimeout(ctx, beaconBlockFetchTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, queryUrl, nil) + if err != nil { + return nil, errServerError + } + resp, err := client.Do(req) + if err != nil { + return nil, errUpstreamError } defer resp.Body.Close() - respObj := &struct { - Data struct { - Message struct { - Body struct { - ExecutionPayload struct { - BlockNumber string `json:"block_number"` - } `json:"execution_payload"` - BlobKzgCommitments []string `json:"blob_kzg_commitments"` - } `json:"body"` - } `json:"message"` - } `json:"data"` - }{} - err = json.NewDecoder(resp.Body).Decode(&respObj) + if resp.StatusCode == http.StatusNotFound { + return nil, errUnknownBlock + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, errUpstreamError + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, maxBeaconBlockResponseSize)) if err != nil { - return 0, nil, errUnknownBlock + return nil, errUpstreamError + } + return parseBeaconBlock(body) +} + +// parseBeaconBlock decodes a /eth/v2/beacon/blocks/{id} JSON body. It is split +// out from the HTTP fetch so that fixtures of every consensus version can be +// tested without a network. +func parseBeaconBlock(body []byte) (*beaconBlock, *httpError) { + respObj := &beaconBlockResponse{} + if err := json.Unmarshal(body, respObj); err != nil { + return nil, errUpstreamError + } + if respObj.Data == nil || respObj.Data.Message == nil || respObj.Data.Message.Body == nil { + return nil, errUpstreamError + } + version := strings.ToLower(strings.TrimSpace(deref(respObj.Version))) + if version == "" { + // the beacon node did not tell us which version this block is + return nil, errUpstreamError } - body := respObj.Data.Message.Body - elBlock, err := strconv.ParseUint(body.ExecutionPayload.BlockNumber, 10, 64) + bodyObj := respObj.Data.Message.Body + slot, hErr := parseSlot(respObj.Data.Message.Slot) + if hErr != nil { + return nil, hErr + } + + var ( + blockHashStr *string + commits *[]string + ) + switch version { + case beaconBlockVersionDeneb, beaconBlockVersionElectra, beaconBlockVersionFulu: + if bodyObj.ExecutionPayload == nil { + return nil, errUpstreamError + } + blockHashStr = bodyObj.ExecutionPayload.BlockHash + commits = bodyObj.BlobKzgCommitments + case beaconBlockVersionGloas: + // Gloas has no execution_payload in the body: the execution block hash + // and the commitments live in the signed bid's message. + if bodyObj.SignedExecutionPayloadBid == nil || bodyObj.SignedExecutionPayloadBid.Message == nil { + return nil, errUpstreamError + } + blockHashStr = bodyObj.SignedExecutionPayloadBid.Message.BlockHash + commits = bodyObj.SignedExecutionPayloadBid.Message.BlobKzgCommitments + default: + return nil, errUnsupportedBeaconVersion + } + + if blockHashStr == nil { + return nil, errUpstreamError + } + executionHash, err := decodeBlockHash(*blockHashStr) if err != nil { - return 0, nil, errUnknownBlock + return nil, errUpstreamError + } + // An empty (or absent) commitments list is valid: the block simply has no blobs. + kzgCommitsAll := []string{} + if commits != nil { + for _, c := range *commits { + if _, err := decodeKZGCommitment(c); err != nil { + return nil, errUpstreamError + } + } + kzgCommitsAll = *commits } - return elBlock, body.BlobKzgCommitments, nil + return &beaconBlock{ + version: version, + slot: slot, + executionHash: executionHash, + kzgCommitsAll: kzgCommitsAll, + }, nil } -// Deprecated since Fusaka -func (a *API) buildSidecar(kvIndex uint64, kzgCommitment []byte, blobHash common.Hash) (*BlobSidecar, *httpError) { +func deref[T any](p *T) T { + var zero T + if p == nil { + return zero + } + return *p +} + +func parseSlot(slot *string) (uint64, *httpError) { + if slot == nil { + return 0, errUpstreamError + } + v, err := strconv.ParseUint(*slot, 10, 64) + if err != nil { + return 0, errUpstreamError + } + return v, nil +} + +func decodeBlockHash(s string) (common.Hash, error) { + b, err := decodeHexFixed(s, common.HashLength) + if err != nil { + return common.Hash{}, err + } + return common.BytesToHash(b), nil +} + +func decodeKZGCommitment(s string) ([]byte, error) { + return decodeHexFixed(s, len(kzg4844.Commitment{})) +} + +func decodeHexFixed(s string, size int) ([]byte, error) { + if !strings.HasPrefix(s, "0x") { + return nil, fmt.Errorf("expected 0x prefixed hex string, got %q", s) + } + b, err := hexutil.Decode(s) + if err != nil { + return nil, err + } + if len(b) != size { + return nil, fmt.Errorf("expected %d bytes, got %d", size, len(b)) + } + return b, nil +} + +// readBlob reads a blob from EthStorage local storage and classifies failures: +// a commit mismatch (the slot holds different data, or nothing at all) means +// the blob is simply not stored here, while an I/O failure is a server error. +func (a *API) readBlob(kvIndex uint64, blobHash common.Hash) ([]byte, *httpError) { blobData, found, err := a.storageMgr.TryRead(kvIndex, int(a.storageMgr.MaxKvSize()), blobHash) if err != nil { + var mismatch *ethstorage.CommitMismatchError + if errors.As(err, &mismatch) { + a.lg.Info("Blob commit mismatch in storage", "kvIndex", kvIndex, "blobHash", blobHash, "err", err) + return nil, errBlobNotInES + } a.lg.Error("Failed to read blob", "err", err) return nil, errServerError } @@ -206,6 +405,15 @@ func (a *API) buildSidecar(kvIndex uint64, kzgCommitment []byte, blobHash common a.lg.Info("Blob not found by storage manager", "kvIndex", kvIndex, "blobHash", blobHash) return nil, errBlobNotInES } + return blobData, nil +} + +// Deprecated since Fusaka +func (a *API) buildSidecar(kvIndex uint64, kzgCommitment []byte, blobHash common.Hash) (*BlobSidecar, *httpError) { + blobData, hErr := a.readBlob(kvIndex, blobHash) + if hErr != nil { + return nil, hErr + } blob := kzg4844.Blob(blobData) kzgProof, err := kzg4844.ComputeBlobProof(&blob, kzg4844.Commitment(kzgCommitment)) if err != nil { diff --git a/ethstorage/archiver/api_e2e_test.go b/ethstorage/archiver/api_e2e_test.go new file mode 100644 index 00000000..0a5a573d --- /dev/null +++ b/ethstorage/archiver/api_e2e_test.go @@ -0,0 +1,455 @@ +// Copyright 2022-2023, EthStorage. +// For license information, see https://github.com/ethstorage/es-node/blob/main/LICENSE + +package archiver + +import ( + "context" + "crypto/sha256" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "math/big" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/crypto/kzg4844" + "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/log" + "github.com/ethstorage/go-ethstorage/ethstorage" + "github.com/ethstorage/go-ethstorage/ethstorage/blobs" + "github.com/ethstorage/go-ethstorage/ethstorage/eth" +) + +// End-to-end tests of the archiver API: a beacon node and an execution client +// are replaced by local fakes, and blobs are written to a real StorageManager +// backed by a temporary shard file. This exercises the whole path: beacon block +// parsing, PutBlob lookup by execution block hash, blob filtering, ordering and +// local reads. + +const ( + e2eKvEntries = uint64(16) + e2eKvSize = uint64(blobs.BlobLength) + e2eChunkSize = uint64(blobs.BlobLength) + e2eSlot = "9999" +) + +var e2eContract = common.HexToAddress("0x000000000000000000000000000000003330001") + +// e2eHarness wires an archiver API to fake beacon and L1 endpoints. +type e2eHarness struct { + api *API + elLogs []types.Log + elErr error + getLogsReq []map[string]any +} + +// makeBlob builds a recognisable blob whose KZG commitment and versioned hash +// are computed the same way the archiver computes them. +func makeBlob(t *testing.T, kvIndex uint64) ([]byte, kzg4844.Commitment, common.Hash) { + t.Helper() + raw := make([]byte, blobs.BlobLength) + copy(raw[:20], e2eContract.Bytes()) + binary.BigEndian.PutUint64(raw[20:28], kvIndex) + // Fill the rest with a per-KV pattern so tests can tell blobs apart. Every + // 32 byte field element stays well below the BLS modulus. + for i := 28; i < len(raw); i++ { + raw[i] = byte(kvIndex) + } + blob := kzg4844.Blob(raw) + commit, err := kzg4844.BlobToCommitment(&blob) + if err != nil { + t.Fatalf("BlobToCommitment failed: %v", err) + } + vh := common.Hash(kzg4844.CalcBlobHashV1(sha256.New(), &commit)) + return raw, commit, vh +} + +// newStorageManager fills kv slots with blobs and returns the manager. +func newStorageManager(t *testing.T, kvIndexes []uint64, blobsData [][]byte, commits []common.Hash) *ethstorage.StorageManager { + t.Helper() + sm := ethstorage.NewShardManager(e2eContract, e2eKvSize, e2eKvEntries, e2eChunkSize) + if err := sm.AddDataShard(0); err != nil { + t.Fatalf("AddDataShard failed: %v", err) + } + fileName := filepath.Join(t.TempDir(), "archiver_e2e.dat") + df, err := ethstorage.Create(fileName, 0, e2eKvEntries*(e2eKvSize/e2eChunkSize), 0, e2eKvSize, ethstorage.NO_ENCODE, common.Address{}, e2eChunkSize) + if err != nil { + t.Fatalf("Create data file failed: %v", err) + } + t.Cleanup(func() { + if err := df.Close(); err != nil { + t.Errorf("close data file failed: %v", err) + } + }) + if err := sm.AddDataFile(df); err != nil { + t.Fatalf("AddDataFile failed: %v", err) + } + + storageMgr := ethstorage.NewStorageManager(sm, &e2eL1Source{kvEntryCount: e2eKvEntries}, log.New("archiver_e2e")) + storageMgr.DownloadThreadNum = 1 + if len(kvIndexes) > 0 { + if err := storageMgr.DownloadFinished(100, kvIndexes, blobsData, commits); err != nil { + t.Fatalf("DownloadFinished failed: %v", err) + } + } + return storageMgr +} + +type e2eL1Source struct { + kvEntryCount uint64 +} + +func (l *e2eL1Source) GetKvMetas(kvIndices []uint64, blockNumber int64) ([][32]byte, error) { + return make([][32]byte, len(kvIndices)), nil +} + +func (l *e2eL1Source) GetStorageKvEntryCount(blockNumber int64) (uint64, error) { + return l.kvEntryCount, nil +} + +// newE2EAPI starts a fake beacon node serving one block of the given version, +// and a fake execution client answering eth_getLogs, and wires them into an API. +func newE2EAPI(t *testing.T, storageMgr *ethstorage.StorageManager, version string, commitments []string, elLogs []types.Log, elErr error) *e2eHarness { + t.Helper() + h := &e2eHarness{ + elLogs: elLogs, + elErr: elErr, + } + + beacon := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/eth/v1/beacon/genesis"): + _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"genesis_time": "1000"}}) + case strings.Contains(r.URL.Path, "/eth/v2/beacon/blocks/"): + _, _ = w.Write([]byte(beaconBlockJSON(version, e2eSlot, testBlockHash, commitments, version == beaconBlockVersionGloas))) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(beacon.Close) + + l1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("bad JSON-RPC request: %v", err) + return + } + resp := map[string]any{"jsonrpc": "2.0", "id": req.ID} + switch req.Method { + case "net_version": + resp["result"] = "1" + case "eth_getLogs": + var query map[string]any + if len(req.Params) > 0 { + _ = json.Unmarshal(req.Params[0], &query) + } + h.getLogsReq = append(h.getLogsReq, query) + if elErr != nil { + resp["error"] = map[string]any{"code": -32000, "message": elErr.Error()} + } else { + resp["result"] = elLogs + } + default: + resp["error"] = map[string]any{"code": -32601, "message": "unsupported in test: " + req.Method} + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(resp) + })) + t.Cleanup(l1.Close) + + beaconClient, err := eth.NewBeaconClient(beacon.URL, 12) + if err != nil { + t.Fatalf("NewBeaconClient failed: %v", err) + } + ec, err := ethclient.Dial(l1.URL) + if err != nil { + t.Fatalf("dial fake L1 failed: %v", err) + } + l1Client := eth.NewClient(context.Background(), ec, true, e2eContract, 1, + func() (*types.Header, error) { return nil, errors.New("head polling disabled in test") }, + log.New("archiver_e2e_l1")) + t.Cleanup(l1Client.Close) + + h.api = NewAPI(storageMgr, beaconClient, l1Client, log.New("archiver_e2e_api")) + return h +} + +// putBlobLog builds the PutBlob log an execution client would return for a blob +// stored at kvIndex with the given versioned hash. +func putBlobLog(kvIndex uint64, blobHash common.Hash, logIndex uint) types.Log { + // PutBlob emits kvIndex in topic 1 and leaves topic 2 empty; the versioned + // blob hash is topic 3. + var emptyTopic common.Hash + topics := []common.Hash{ + crypto.Keccak256Hash([]byte(eth.PutBlobEvent)), + common.BigToHash(big.NewInt(int64(kvIndex))), + emptyTopic, + blobHash, + } + return types.Log{ + Address: e2eContract, + Topics: topics, + Data: []byte{}, + BlockNumber: 100, + BlockHash: common.HexToHash(testBlockHash), + TxHash: common.Hash{byte(logIndex) + 1}, + Index: logIndex, + } +} + +// TestQueryBlobsOrderAndFilter stores three blobs at shuffled kv indices and +// checks that queryBlobs returns them in beacon commitment order, filtered to +// the requested versioned hashes. +func TestQueryBlobsOrderAndFilter(t *testing.T) { + // commitment order -> kv index mapping is deliberately shuffled + kvIndexes := []uint64{5, 1, 3} + raw := make([][]byte, 3) + commits := make([]kzg4844.Commitment, 3) + hashes := make([]common.Hash, 3) + for i, kv := range kvIndexes { + raw[i], commits[i], hashes[i] = makeBlob(t, kv) + } + + commitHex := make([]string, 3) + for i := range commits { + commitHex[i] = "0x" + common.Bytes2Hex(commits[i][:]) + } + + // The execution client reports the logs in an order unrelated to the + // commitment order. + elLogs := []types.Log{ + putBlobLog(kvIndexes[1], hashes[1], 0), + putBlobLog(kvIndexes[2], hashes[2], 1), + putBlobLog(kvIndexes[0], hashes[0], 2), + } + + storageMgr := newStorageManager(t, kvIndexes, raw, hashes) + h := newE2EAPI(t, storageMgr, beaconBlockVersionGloas, commitHex, elLogs, nil) + + // ask for blob 2 and blob 0 only, out of order + res, hErr := h.api.queryBlobs(context.Background(), e2eSlot, []common.Hash{hashes[2], hashes[0]}) + if hErr != nil { + t.Fatalf("queryBlobs failed: %d %s", hErr.Code, hErr.Message) + } + if len(res.Data) != 2 { + t.Fatalf("got %d blobs, want 2", len(res.Data)) + } + // commitment order wins over both the request order and the EL log order + wantOrder := []int{0, 2} + for n, want := range wantOrder { + got, err := hexToBlob(res.Data[n]) + if err != nil { + t.Fatalf("blob %d not decodable: %v", n, err) + } + if !equalBytes(got, raw[want]) { + t.Errorf("blob %d is kv %d, want kv %d", n, blobKvIndex(got), kvIndexes[want]) + } + } + + // the archiver must query the execution block by hash, not by number + assertGetLogsByHash(t, h.getLogsReq) +} + +// TestQueryBlobSidecarsGloas checks the deprecated sidecar endpoint on a Gloas +// block: it must resolve the execution block from the bid and return sidecars +// with the commitment and proof matching the block. +func TestQueryBlobSidecarsGloas(t *testing.T) { + kvIndexes := []uint64{2, 7} + raw := make([][]byte, 2) + commits := make([]kzg4844.Commitment, 2) + hashes := make([]common.Hash, 2) + for i, kv := range kvIndexes { + raw[i], commits[i], hashes[i] = makeBlob(t, kv) + } + commitHex := []string{"0x" + common.Bytes2Hex(commits[0][:]), "0x" + common.Bytes2Hex(commits[1][:])} + elLogs := []types.Log{putBlobLog(kvIndexes[0], hashes[0], 0), putBlobLog(kvIndexes[1], hashes[1], 1)} + + storageMgr := newStorageManager(t, kvIndexes, raw, hashes) + h := newE2EAPI(t, storageMgr, beaconBlockVersionGloas, commitHex, elLogs, nil) + + res, hErr := h.api.queryBlobSidecars(context.Background(), e2eSlot, []uint64{1}) + if hErr != nil { + t.Fatalf("queryBlobSidecars failed: %d %s", hErr.Code, hErr.Message) + } + if len(res.Data) != 1 { + t.Fatalf("got %d sidecars, want 1", len(res.Data)) + } + sidecar := res.Data[0] + if sidecar.Index != 1 { + t.Errorf("sidecar index = %d, want 1", sidecar.Index) + } + if !equalBytes(sidecar.KZGCommitment[:], commits[1][:]) { + t.Error("sidecar commitment does not match the block commitment") + } + if !equalBytes(sidecar.Blob[:], raw[1]) { + t.Error("sidecar blob does not match stored data") + } + sidecarBlob := kzg4844.Blob(raw[1]) + proof, err := kzg4844.ComputeBlobProof(&sidecarBlob, commits[1]) + if err != nil { + t.Fatal(err) + } + if !equalBytes(sidecar.KZGProof[:], proof[:]) { + t.Error("sidecar proof mismatch") + } + assertGetLogsByHash(t, h.getLogsReq) +} + +// TestQueryBlobsStorageMissing covers a block that exists on both the beacon +// chain and L1, but whose blob is not stored locally. +func TestQueryBlobsStorageMissing(t *testing.T) { + kv := uint64(4) + raw, commit, hash := makeBlob(t, kv) + commitHex := []string{"0x" + common.Bytes2Hex(commit[:])} + elLogs := []types.Log{putBlobLog(kv, hash, 0)} + + // storage manager exists but nothing was written into it + storageMgr := newStorageManager(t, nil, nil, nil) + h := newE2EAPI(t, storageMgr, beaconBlockVersionDeneb, commitHex, elLogs, nil) + + _, hErr := h.api.queryBlobs(context.Background(), e2eSlot, []common.Hash{hash}) + if hErr == nil { + t.Fatal("expected an error for a blob that is not stored") + } + if hErr.Code != http.StatusNotFound || hErr.Message != errBlobNotInES.Message { + t.Errorf("error = %d %q, want 404 %q", hErr.Code, hErr.Message, errBlobNotInES.Message) + } + _ = raw +} + +// TestQueryBlobsExecutionBlockNotAvailable covers a Gloas bid whose execution +// block is not known to the L1 client yet: the request must be retryable, not a +// final "block not found". +func TestQueryBlobsExecutionBlockNotAvailable(t *testing.T) { + kv := uint64(4) + _, commit, hash := makeBlob(t, kv) + commitHex := []string{"0x" + common.Bytes2Hex(commit[:])} + + storageMgr := newStorageManager(t, nil, nil, nil) + h := newE2EAPI(t, storageMgr, beaconBlockVersionGloas, commitHex, nil, errors.New("unknown block")) + + _, hErr := h.api.queryBlobs(context.Background(), e2eSlot, []common.Hash{hash}) + if hErr == nil { + t.Fatal("expected an error when the execution block is unavailable") + } + if hErr.Code != http.StatusServiceUnavailable { + t.Errorf("error = %d %q, want %d %q", hErr.Code, hErr.Message, + errExecutionBlockNotAvailable.Code, errExecutionBlockNotAvailable.Message) + } + if hErr.Code == http.StatusNotFound { + t.Error("an unavailable execution block must not be reported as a missing block") + } +} + +// TestQueryBlobsBeaconBlockNotFound covers a slot the beacon node does not know. +func TestQueryBlobsBeaconBlockNotFound(t *testing.T) { + storageMgr := newStorageManager(t, nil, nil, nil) + h := newE2EAPI(t, storageMgr, beaconBlockVersionDeneb, nil, nil, nil) + // make the beacon answer 404 for the block + h.api.beaconClient = newNotFoundBeaconClient(t) + + _, hErr := h.api.queryBlobs(context.Background(), "123456", nil) + if hErr == nil { + t.Fatal("expected an error for an unknown slot") + } + if hErr.Code != http.StatusNotFound || hErr.Message != errUnknownBlock.Message { + t.Errorf("error = %d %q, want 404 %q", hErr.Code, hErr.Message, errUnknownBlock.Message) + } +} + +// TestQueryBlobsUnsupportedVersion makes sure a block the archiver cannot +// decode is never reported as "block not found". +func TestQueryBlobsUnsupportedVersion(t *testing.T) { + storageMgr := newStorageManager(t, nil, nil, nil) + h := newE2EAPI(t, storageMgr, "prague", nil, nil, nil) + + _, hErr := h.api.queryBlobs(context.Background(), e2eSlot, nil) + if hErr == nil { + t.Fatal("expected an error for an unsupported beacon version") + } + if hErr.Code != errUnsupportedBeaconVersion.Code { + t.Errorf("error = %d %q, want %d %q", hErr.Code, hErr.Message, + errUnsupportedBeaconVersion.Code, errUnsupportedBeaconVersion.Message) + } +} + +func newNotFoundBeaconClient(t *testing.T) *eth.BeaconClient { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/genesis") { + _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"genesis_time": "1000"}}) + return + } + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"code":404,"message":"Block not found"}`)) + })) + t.Cleanup(srv.Close) + client, err := eth.NewBeaconClient(srv.URL, 12) + if err != nil { + t.Fatalf("NewBeaconClient failed: %v", err) + } + return client +} + +func assertGetLogsByHash(t *testing.T, requests []map[string]any) { + t.Helper() + if len(requests) != 1 { + t.Fatalf("expected exactly one eth_getLogs call, got %d", len(requests)) + } + query := requests[0] + if query["blockHash"] != common.HexToHash(testBlockHash).Hex() { + t.Errorf("eth_getLogs blockHash = %v, want %s", query["blockHash"], testBlockHash) + } + for _, forbidden := range []string{"fromBlock", "toBlock"} { + if _, ok := query[forbidden]; ok { + t.Errorf("eth_getLogs must not set %s when querying by block hash", forbidden) + } + } + topic := crypto.Keccak256Hash([]byte(eth.PutBlobEvent)).Hex() + topics, ok := query["topics"].([]any) + if !ok || len(topics) != 1 { + t.Fatalf("eth_getLogs topics = %v, want a single topic list", query["topics"]) + } + first, ok := topics[0].([]any) + if !ok || len(first) != 1 || first[0] != topic { + t.Errorf("eth_getLogs topics[0] = %v, want [%s]", topics[0], topic) + } +} + +func hexToBlob(s string) ([]byte, error) { + if !strings.HasPrefix(s, "0x") { + return nil, fmt.Errorf("expected 0x prefixed hex, got %d chars", len(s)) + } + return common.FromHex(s), nil +} + +func blobKvIndex(blob []byte) uint64 { + if len(blob) < 28 { + return 0 + } + return binary.BigEndian.Uint64(blob[20:28]) +} + +func equalBytes(a, b []byte) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/ethstorage/archiver/api_parse_test.go b/ethstorage/archiver/api_parse_test.go new file mode 100644 index 00000000..0c1c30c3 --- /dev/null +++ b/ethstorage/archiver/api_parse_test.go @@ -0,0 +1,307 @@ +// Copyright 2022-2023, EthStorage. +// For license information, see https://github.com/ethstorage/es-node/blob/main/LICENSE + +package archiver + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/ethereum/go-ethereum/common" +) + +// A valid 32 byte execution block hash, plus 48 byte KZG commitments built from +// a seed byte so their length is guaranteed. +const testBlockHash = "0x1111111111111111111111111111111111111111111111111111111111111111" + +func commitmentHex(seed byte) string { + b := make([]byte, 48) + for i := range b { + b[i] = seed + byte(i) + } + return "0x" + common.Bytes2Hex(b) +} + +var ( + testCommit0 = commitmentHex(0x89) + testCommit1 = commitmentHex(0xa1) + testCommit2 = commitmentHex(0x90) +) + +// beaconBlockJSON builds a /eth/v2/beacon/blocks response body the way a beacon +// node would serve it for the given consensus version. +func beaconBlockJSON(version, slot, elBlockHash string, commitments []string, useBid bool) string { + commitsJSON := "null" + if commitments != nil { + quoted := make([]string, len(commitments)) + for i, c := range commitments { + quoted[i] = fmt.Sprintf("%q", c) + } + commitsJSON = "[" + strings.Join(quoted, ",") + "]" + } + + var body string + if useBid { + body = fmt.Sprintf(`{"signed_execution_payload_bid":{"message":{`+ + `"block_hash":%q,"blob_kzg_commitments":%s}}}`, elBlockHash, commitsJSON) + } else { + body = fmt.Sprintf(`{"execution_payload":{"block_hash":%q},"blob_kzg_commitments":%s}`, + elBlockHash, commitsJSON) + } + return fmt.Sprintf(`{"version":%q,"data":{"message":{"slot":%q,"body":%s}}}`, version, slot, body) +} + +func TestParseBeaconBlockVersions(t *testing.T) { + cases := []struct { + name string + version string + useBid bool + commitments []string + }{ + {"deneb", beaconBlockVersionDeneb, false, []string{testCommit0}}, + {"electra", beaconBlockVersionElectra, false, []string{testCommit0, testCommit1, testCommit2}}, + {"fulu", beaconBlockVersionFulu, false, []string{testCommit1}}, + // Gloas blocks have no execution_payload at all: everything lives in + // the signed bid. + {"gloas", beaconBlockVersionGloas, true, []string{testCommit0, testCommit1}}, + {"gloas without blobs", beaconBlockVersionGloas, true, []string{}}, + {"deneb without blobs", beaconBlockVersionDeneb, false, nil}, + {"version case insensitive", "Gloas", true, []string{testCommit0}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + body := beaconBlockJSON(tc.version, "1234", testBlockHash, tc.commitments, tc.useBid) + block, hErr := parseBeaconBlock([]byte(body)) + if hErr != nil { + t.Fatalf("parseBeaconBlock failed: %d %s", hErr.Code, hErr.Message) + } + if block.executionHash != common.HexToHash(testBlockHash) { + t.Errorf("executionHash = %s, want %s", block.executionHash, testBlockHash) + } + if block.slot != 1234 { + t.Errorf("slot = %d, want 1234", block.slot) + } + want := tc.commitments + if want == nil { + want = []string{} + } + if len(block.kzgCommitsAll) != len(want) { + t.Fatalf("got %d commitments, want %d", len(block.kzgCommitsAll), len(want)) + } + // Commitments must keep the order they appear in the block: the + // archiver relies on it to map blobs to their index. + for i, c := range want { + if block.kzgCommitsAll[i] != c { + t.Errorf("commitment %d = %s, want %s", i, block.kzgCommitsAll[i], c) + } + } + }) + } +} + +func TestParseBeaconBlockErrors(t *testing.T) { + cases := []struct { + name string + body string + want *httpError + }{ + { + name: "unknown version", + body: beaconBlockJSON("prague", "1", testBlockHash, nil, false), + want: errUnsupportedBeaconVersion, + }, + { + name: "missing version", + body: `{"data":{"message":{"slot":"1","body":{"execution_payload":{"block_hash":"` + testBlockHash + `"}}}}}`, + want: errUpstreamError, + }, + { + name: "deneb without execution payload", + body: beaconBlockJSON("deneb", "1", testBlockHash, []string{testCommit0}, true), + want: errUpstreamError, + }, + { + name: "gloas without signed bid", + body: beaconBlockJSON("gloas", "1", testBlockHash, []string{testCommit0}, false), + want: errUpstreamError, + }, + { + name: "gloas bid without block hash", + body: `{"version":"gloas","data":{"message":{"slot":"1","body":{"signed_execution_payload_bid":{"message":` + + `{"blob_kzg_commitments":[` + fmt.Sprintf("%q", testCommit0) + `]}}}}}}`, + want: errUpstreamError, + }, + { + name: "execution payload without block hash", + body: `{"version":"deneb","data":{"message":{"slot":"1","body":{"blob_kzg_commitments":null}}}}`, + want: errUpstreamError, + }, + { + name: "missing slot", + body: `{"version":"deneb","data":{"message":{"body":{"execution_payload":{"block_hash":"` + testBlockHash + `"}}}}}`, + want: errUpstreamError, + }, + { + name: "block hash too short", + body: beaconBlockJSON("deneb", "1", "0x111111111111111111111111111111111111111111111111111111111111", nil, false), + want: errUpstreamError, + }, + { + name: "commitment too short", + body: beaconBlockJSON("deneb", "1", testBlockHash, []string{"0x11111111111111111111"}, false), + want: errUpstreamError, + }, + { + name: "commitment not hex", + body: beaconBlockJSON("deneb", "1", testBlockHash, []string{"notahex"}, false), + want: errUpstreamError, + }, + { + name: "invalid json", + body: `{"version":"deneb",`, + want: errUpstreamError, + }, + { + name: "empty body", + body: ``, + want: errUpstreamError, + }, + { + name: "null data", + body: `{"version":"deneb","data":null}`, + want: errUpstreamError, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + block, hErr := parseBeaconBlock([]byte(tc.body)) + if hErr == nil { + t.Fatalf("expected an error, got block %+v", block) + } + if block != nil { + t.Errorf("expected no block alongside the error") + } + if hErr.Code != tc.want.Code || hErr.Message != tc.want.Message { + t.Errorf("error = %d %q, want %d %q", hErr.Code, hErr.Message, tc.want.Code, tc.want.Message) + } + // A malformed or unsupported block must never look like a block + // that does not exist: callers would stop retrying. + if hErr.Code == http.StatusNotFound { + t.Errorf("classification error: %d %q must not be 404", hErr.Code, hErr.Message) + } + }) + } +} + +func TestFetchBeaconBlockStatusCodes(t *testing.T) { + cases := []struct { + name string + status int + body string + want *httpError + }{ + { + name: "beacon 404 is block not found", + status: http.StatusNotFound, + body: `{"code":404,"message":"Block not found"}`, + want: errUnknownBlock, + }, + { + name: "beacon 500 is upstream error", + status: http.StatusInternalServerError, + body: `{"code":500,"message":"Internal Server Error"}`, + want: errUpstreamError, + }, + { + name: "beacon 503 is upstream error", + status: http.StatusServiceUnavailable, + body: `{"code":503,"message":"beacon node is syncing"}`, + want: errUpstreamError, + }, + { + name: "beacon 200 with garbage is upstream error", + status: http.StatusOK, + body: `proxy error`, + want: errUpstreamError, + }, + { + name: "beacon 200 with unsupported version", + status: http.StatusOK, + body: beaconBlockJSON("prague", "1", testBlockHash, nil, false), + want: errUnsupportedBeaconVersion, + }, + { + name: "beacon 200 gloas block", + status: http.StatusOK, + body: beaconBlockJSON("gloas", "9", testBlockHash, []string{testCommit0}, true), + want: nil, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(tc.body)) + })) + defer srv.Close() + + block, hErr := fetchBeaconBlock(t.Context(), srv.Client(), srv.URL) + if tc.want == nil { + if hErr != nil { + t.Fatalf("unexpected error: %d %s", hErr.Code, hErr.Message) + } + if block.executionHash != common.HexToHash(testBlockHash) { + t.Errorf("executionHash = %s, want %s", block.executionHash, testBlockHash) + } + return + } + if hErr == nil { + t.Fatalf("expected %d %q, got block %+v", tc.want.Code, tc.want.Message, block) + } + if hErr.Code != tc.want.Code || hErr.Message != tc.want.Message { + t.Errorf("error = %d %q, want %d %q", hErr.Code, hErr.Message, tc.want.Code, tc.want.Message) + } + }) + } +} + +func TestIsUnknownBlockErr(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {err: fmt.Errorf("unknown block"), want: true}, + {err: fmt.Errorf(`{"code":-32000,"message":"unknown block"}`), want: true}, + {err: fmt.Errorf("Block not found"), want: true}, + {err: fmt.Errorf("connection refused"), want: false}, + {err: fmt.Errorf("context deadline exceeded"), want: false}, + {err: nil, want: false}, + } + for _, tc := range cases { + if got := isUnknownBlockErr(tc.err); got != tc.want { + t.Errorf("isUnknownBlockErr(%v) = %v, want %v", tc.err, got, tc.want) + } + } +} + +// TestBeaconBlockResponseShape makes sure the decode struct keeps up with the +// JSON the beacon nodes actually serve for each version. +func TestBeaconBlockResponseShape(t *testing.T) { + var resp beaconBlockResponse + if err := json.Unmarshal([]byte(beaconBlockJSON("gloas", "7", testBlockHash, []string{testCommit0}, true)), &resp); err != nil { + t.Fatal(err) + } + if resp.Data.Message.Body.ExecutionPayload != nil { + t.Error("gloas body must not decode an execution payload") + } + if resp.Data.Message.Body.SignedExecutionPayloadBid == nil { + t.Fatal("gloas bid not decoded") + } + if *resp.Data.Message.Body.SignedExecutionPayloadBid.Message.BlockHash != testBlockHash { + t.Error("bid block hash not decoded") + } +} diff --git a/ethstorage/archiver/service.go b/ethstorage/archiver/service.go index 3d242652..3842801a 100644 --- a/ethstorage/archiver/service.go +++ b/ethstorage/archiver/service.go @@ -52,7 +52,7 @@ func (a *APIService) blobSidecarHandler(w http.ResponseWriter, r *http.Request) return } - result, hErr := a.api.queryBlobSidecars(id, indices) + result, hErr := a.api.queryBlobSidecars(r.Context(), id, indices) if hErr != nil { hErr.write(w) return @@ -90,7 +90,7 @@ func (a *APIService) blobsHandler(w http.ResponseWriter, r *http.Request) { return } - result, hErr := a.api.queryBlobs(id, hashes) + result, hErr := a.api.queryBlobs(r.Context(), id, hashes) if hErr != nil { hErr.write(w) return diff --git a/ethstorage/archiver/utils.go b/ethstorage/archiver/utils.go index f477cb2b..5211ecfc 100644 --- a/ethstorage/archiver/utils.go +++ b/ethstorage/archiver/utils.go @@ -76,8 +76,40 @@ var ( Code: http.StatusInternalServerError, Message: "Internal server error", } + // errUpstreamError covers beacon node transport errors, timeouts, 5xx + // responses, invalid JSON, and blocks missing required fields. It is a 5xx + // so that callers (e.g. op-node) retry instead of treating it as "block + // not found". + errUpstreamError = &httpError{ + Code: http.StatusBadGateway, + Message: "Upstream beacon node error", + } + // errUnsupportedBeaconVersion is returned when the beacon node reports a + // version this archiver cannot decode. + errUnsupportedBeaconVersion = &httpError{ + Code: http.StatusBadGateway, + Message: "Unsupported beacon block version", + } + // errExecutionBlockNotAvailable is returned when the beacon block resolved + // fine but the execution block it points to is not known to the L1 client + // yet (e.g. a Gloas bid whose payload has not arrived). Retryable 5xx. + errExecutionBlockNotAvailable = &httpError{ + Code: http.StatusServiceUnavailable, + Message: "Execution block not available on L1 yet", + } ) +// isUnknownBlockErr reports whether an eth_getLogs error means the queried +// block hash is unknown to the execution client (geth answers with +// "unknown block"), as opposed to a transport failure. +func isUnknownBlockErr(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "not found") || strings.Contains(msg, "unknown block") +} + func newBadRequestError(input string) *httpError { return &httpError{ Code: http.StatusBadRequest, diff --git a/ethstorage/eth/beacon_client.go b/ethstorage/eth/beacon_client.go index cd457327..458e6c0b 100644 --- a/ethstorage/eth/beacon_client.go +++ b/ethstorage/eth/beacon_client.go @@ -10,6 +10,7 @@ import ( "net/http" "net/url" "strconv" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethstorage/go-ethstorage/ethstorage/blobs" @@ -19,6 +20,7 @@ type BeaconClient struct { beaconURL string genesisSlotTime uint64 slotTime uint64 + httpClient *http.Client } type Blob struct { @@ -35,6 +37,7 @@ func NewBeaconClient(url string, slotTime uint64) (*BeaconClient, error) { beaconURL: url, genesisSlotTime: genesisSlotTime, slotTime: slotTime, + httpClient: &http.Client{Timeout: 30 * time.Second}, } return res, nil } @@ -75,7 +78,7 @@ func (c *BeaconClient) DownloadBlobs(timestamp uint64, hashes []common.Hash) (ma if err != nil { return nil, err } - resp, err := http.Get(blobsUrl) + resp, err := c.httpClient.Get(blobsUrl) if err != nil { return nil, fmt.Errorf("failed to query beacon blobs with url %s: %w", blobsUrl, err) } @@ -125,3 +128,13 @@ func (c *BeaconClient) QueryUrlForV1BeaconBlobs(timestamp uint64, hashes []commo } return blobsURL + "?" + q.Encode(), nil } + +// HTTPClient returns the HTTP client used to query the beacon node, so that +// callers can issue requests with their own context (cancellation and +// timeouts) instead of fire-and-forget http.Get. +func (c *BeaconClient) HTTPClient() *http.Client { + if c.httpClient != nil { + return c.httpClient + } + return http.DefaultClient +} diff --git a/ethstorage/eth/polling_client.go b/ethstorage/eth/polling_client.go index ba95e512..63b607fb 100644 --- a/ethstorage/eth/polling_client.go +++ b/ethstorage/eth/polling_client.go @@ -245,6 +245,28 @@ func (w *PollingClient) FilterLogsByBlockRange(start *big.Int, end *big.Int, eve return w.FilterLogs(context.Background(), query) } +// FilterLogsByBlockHash retrieves events emitted in the block with the given +// hash (EIP-234). blockHash is mutually exclusive with fromBlock/toBlock, so +// they must not be set together. This keeps log queries aligned with a +// specific block even during reorgs. +func (w *PollingClient) FilterLogsByBlockHash(ctx context.Context, blockHash common.Hash, eventSig string) ([]types.Log, error) { + topic := crypto.Keccak256Hash([]byte(eventSig)) + + // create a new filter query + query := ethereum.FilterQuery{ + Addresses: []common.Address{w.esContract}, + Topics: [][]common.Hash{ + { + topic, + }, + }, + BlockHash: &blockHash, + } + + // retrieve events that match the filter query + return w.FilterLogs(ctx, query) +} + func (w *PollingClient) GetStorageKvEntryCount(blockNumber int64) (uint64, error) { h := crypto.Keccak256Hash([]byte(`kvEntryCount()`)) diff --git a/ethstorage/eth/polling_client_test.go b/ethstorage/eth/polling_client_test.go new file mode 100644 index 00000000..72f7e62d --- /dev/null +++ b/ethstorage/eth/polling_client_test.go @@ -0,0 +1,221 @@ +// Copyright 2022-2023, EthStorage. +// For license information, see https://github.com/ethstorage/es-node/blob/main/LICENSE + +package eth + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/log" +) + +var testContract = common.HexToAddress("0x000000000000000000000000000000003330001") + +// fakeRPC is a minimal JSON-RPC server that records every request it serves so +// that tests can assert on the exact parameters the client sent. +type fakeRPC struct { + server *httptest.Server + mu sync.Mutex + requests []rpcRequest +} + +type rpcRequest struct { + Method string `json:"-"` + Params []json.RawMessage `json:"-"` +} + +func (f *fakeRPC) byMethod(method string) []rpcRequest { + f.mu.Lock() + defer f.mu.Unlock() + var out []rpcRequest + for _, req := range f.requests { + if req.Method == method { + out = append(out, req) + } + } + return out +} + +func newFakeRPC(t *testing.T, handle func(method string, params []json.RawMessage) (any, error)) *fakeRPC { + t.Helper() + f := &fakeRPC{} + f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("failed to decode JSON-RPC request: %v", err) + return + } + f.mu.Lock() + f.requests = append(f.requests, rpcRequest{Method: req.Method, Params: req.Params}) + f.mu.Unlock() + + var res any + var err error + if req.Method == "net_version" { + // PollingClient.NewClient fails hard without a usable network id. + res = "1" + } else { + res, err = handle(req.Method, req.Params) + } + resp := map[string]any{"jsonrpc": "2.0", "id": req.ID} + if err != nil { + resp["error"] = map[string]any{"code": -32000, "message": err.Error()} + } else { + resp["result"] = res + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + t.Errorf("failed to encode JSON-RPC response: %v", err) + } + })) + t.Cleanup(f.server.Close) + return f +} + +// newTestPollingClient builds a PollingClient against a fake RPC endpoint. The +// head poller is given a stub queryHeader so tests do not have to emulate +// eth_getBlockByNumber. pollRate must be non-zero: the poll loop only re-arms +// (and so only reacts to context cancellation on Close) when a next poll is +// scheduled. +func newTestPollingClient(t *testing.T, rpc *fakeRPC) *PollingClient { + t.Helper() + ec, err := ethclient.Dial(rpc.server.URL) + if err != nil { + t.Fatalf("failed to dial fake RPC: %v", err) + } + client := NewClient( + context.Background(), ec, true, testContract, 1, + func() (*types.Header, error) { return nil, errors.New("head polling disabled in test") }, + log.New("polling_client_test"), + ) + t.Cleanup(client.Close) + return client +} + +// TestFilterLogsByBlockHashQuery asserts the eth_getLogs request built by +// FilterLogsByBlockHash: it filters on the EthStorage contract and the event +// topic, queries the block by hash, and (per EIP-234) never combines blockHash +// with fromBlock/toBlock. +func TestFilterLogsByBlockHashQuery(t *testing.T) { + blockHash := common.HexToHash("0x2c8c22f8ba515dcf791d6100936e7e36168575a0e604acc0ddab4c6c0f9f0f4c") + topic := crypto.Keccak256Hash([]byte(PutBlobEvent)) + + rpc := newFakeRPC(t, func(method string, params []json.RawMessage) (any, error) { + return []any{}, nil + }) + client := newTestPollingClient(t, rpc) + + logs, err := client.FilterLogsByBlockHash(context.Background(), blockHash, PutBlobEvent) + if err != nil { + t.Fatalf("FilterLogsByBlockHash failed: %v", err) + } + if len(logs) != 0 { + t.Fatalf("expected no logs, got %d", len(logs)) + } + + calls := rpc.byMethod("eth_getLogs") + if len(calls) != 1 { + t.Fatalf("expected exactly one eth_getLogs call, got %d", len(calls)) + } + if len(calls[0].Params) != 1 { + t.Fatalf("expected a single filter argument, got %d", len(calls[0].Params)) + } + + var query map[string]any + if err := json.Unmarshal(calls[0].Params[0], &query); err != nil { + t.Fatalf("failed to decode filter argument: %v", err) + } + + if got, want := query["blockHash"], blockHash.Hex(); got != want { + t.Errorf("blockHash = %v, want %v", got, want) + } + // EIP-234: blockHash and fromBlock/toBlock are mutually exclusive. A server + // that received both would reject the request outright. + for _, forbidden := range []string{"fromBlock", "toBlock"} { + if _, ok := query[forbidden]; ok { + t.Errorf("filter must not set %s together with blockHash: %v", forbidden, query) + } + } + + addr, ok := query["address"].([]any) + if !ok || len(addr) != 1 || addr[0] != testContract.Hex() { + t.Errorf("address = %v, want [%s]", query["address"], testContract.Hex()) + } + + topics, ok := query["topics"].([]any) + if !ok || len(topics) != 1 { + t.Fatalf("topics = %v, want a single topic list", query["topics"]) + } + first, ok := topics[0].([]any) + if !ok || len(first) != 1 || first[0] != topic.Hex() { + t.Errorf("topics[0] = %v, want [%s]", topics[0], topic.Hex()) + } +} + +// TestFilterLogsByBlockHashPropagatesUnknownBlock makes sure the "unknown +// block" answer an execution client gives for an unknown block hash reaches +// the caller verbatim, so the archiver can tell it apart from a transport +// failure. +func TestFilterLogsByBlockHashPropagatesUnknownBlock(t *testing.T) { + rpc := newFakeRPC(t, func(method string, params []json.RawMessage) (any, error) { + return nil, errors.New("unknown block") + }) + client := newTestPollingClient(t, rpc) + + _, err := client.FilterLogsByBlockHash(context.Background(), common.Hash{1}, PutBlobEvent) + if err == nil { + t.Fatal("expected an error for an unknown block hash") + } + if got := err.Error(); got != "unknown block" { + t.Errorf("error = %q, want %q", got, "unknown block") + } +} + +// TestFilterLogsByBlockRangeStillUsesBlockRange guards the existing range +// query against the blockHash change: it must keep sending fromBlock/toBlock +// and no blockHash. +func TestFilterLogsByBlockRangeStillUsesBlockRange(t *testing.T) { + rpc := newFakeRPC(t, func(method string, params []json.RawMessage) (any, error) { + return []any{}, nil + }) + client := newTestPollingClient(t, rpc) + + if _, err := client.FilterLogsByBlockRange(big.NewInt(1), big.NewInt(2), PutBlobEvent); err != nil { + t.Fatalf("FilterLogsByBlockRange failed: %v", err) + } + + calls := rpc.byMethod("eth_getLogs") + if len(calls) != 1 { + t.Fatalf("expected exactly one eth_getLogs call, got %d", len(calls)) + } + var query map[string]any + if err := json.Unmarshal(calls[0].Params[0], &query); err != nil { + t.Fatalf("failed to decode filter argument: %v", err) + } + if _, ok := query["blockHash"]; ok { + t.Errorf("range query must not set blockHash: %v", query) + } + if query["fromBlock"] != "0x1" || query["toBlock"] != "0x2" { + t.Errorf("fromBlock/toBlock = %v/%v, want 0x1/0x2", query["fromBlock"], query["toBlock"]) + } + if fmt.Sprint(query["address"]) != fmt.Sprint([]any{testContract.Hex()}) { + t.Errorf("address = %v, want %s", query["address"], testContract.Hex()) + } +} From b557c96145d6d4a72bbd7cb8fba182cb2693c455 Mon Sep 17 00:00:00 2001 From: syntrust Date: Sat, 10 Oct 2026 18:44:58 +0800 Subject: [PATCH 2/4] archiver: require blob_kzg_commitments and drop downloader client timeout blob_kzg_commitments is required by the beacon API schema: a block without blobs serves an explicit empty array. A missing or null field is now an upstream error instead of being read as 'no blobs', which used to surface as a misleading 404 even when the blob is stored locally. Revert the BeaconClient HTTP client change: DownloadBlobs is also used by the regular node downloader, and a fixed 30s client timeout would fail downloads that previously completed later. fetchBeaconBlock already bounds its own request with a context timeout. --- ethstorage/archiver/api.go | 23 ++++++++----- ethstorage/archiver/api_parse_test.go | 49 +++++++++++++++++++++++---- ethstorage/eth/beacon_client.go | 15 +------- 3 files changed, 57 insertions(+), 30 deletions(-) diff --git a/ethstorage/archiver/api.go b/ethstorage/archiver/api.go index 2884d4ce..d23bdd45 100644 --- a/ethstorage/archiver/api.go +++ b/ethstorage/archiver/api.go @@ -209,7 +209,9 @@ func (a *API) queryBeaconBlock(ctx context.Context, id string) (*beaconBlock, *h a.lg.Error("Invalid beaconID", "beaconID", id, "err", err) return nil, errUnknownBlock } - block, hErr := fetchBeaconBlock(ctx, a.beaconClient.HTTPClient(), queryUrl) + // fetchBeaconBlock bounds the request with its own context timeout, so the + // shared client needs no timeout of its own. + block, hErr := fetchBeaconBlock(ctx, http.DefaultClient, queryUrl) if hErr != nil { a.lg.Error("Failed to get beacon block", "beaconID", id, "err", hErr) return nil, hErr @@ -324,15 +326,18 @@ func parseBeaconBlock(body []byte) (*beaconBlock, *httpError) { if err != nil { return nil, errUpstreamError } - // An empty (or absent) commitments list is valid: the block simply has no blobs. - kzgCommitsAll := []string{} - if commits != nil { - for _, c := range *commits { - if _, err := decodeKZGCommitment(c); err != nil { - return nil, errUpstreamError - } + // blob_kzg_commitments is required by the beacon API schema for every + // supported version: a block without blobs serves an explicit empty array, + // so a missing or null field means the upstream response is malformed. + // Treating it as "no blobs" would report a valid stored blob as not found. + if commits == nil { + return nil, errUpstreamError + } + kzgCommitsAll := *commits + for _, c := range kzgCommitsAll { + if _, err := decodeKZGCommitment(c); err != nil { + return nil, errUpstreamError } - kzgCommitsAll = *commits } return &beaconBlock{ version: version, diff --git a/ethstorage/archiver/api_parse_test.go b/ethstorage/archiver/api_parse_test.go index 0c1c30c3..83367707 100644 --- a/ethstorage/archiver/api_parse_test.go +++ b/ethstorage/archiver/api_parse_test.go @@ -32,11 +32,18 @@ var ( testCommit2 = commitmentHex(0x90) ) +// omitCommitments is passed to beaconBlockJSON to leave blob_kzg_commitments +// out of the response body entirely, as opposed to serving it as null. +var omitCommitments = []string{"\x00omit"} + // beaconBlockJSON builds a /eth/v2/beacon/blocks response body the way a beacon -// node would serve it for the given consensus version. +// node would serve it for the given consensus version. A nil commitments slice +// emits JSON null, omitCommitments leaves the field out, and []string{} emits +// an explicit empty array. func beaconBlockJSON(version, slot, elBlockHash string, commitments []string, useBid bool) string { + omit := commitments != nil && len(commitments) == 1 && commitments[0] == omitCommitments[0] commitsJSON := "null" - if commitments != nil { + if commitments != nil && !omit { quoted := make([]string, len(commitments)) for i, c := range commitments { quoted[i] = fmt.Sprintf("%q", c) @@ -44,13 +51,17 @@ func beaconBlockJSON(version, slot, elBlockHash string, commitments []string, us commitsJSON = "[" + strings.Join(quoted, ",") + "]" } + commitmentsField := fmt.Sprintf(`,"blob_kzg_commitments":%s`, commitsJSON) + if omit { + commitmentsField = "" + } + var body string if useBid { - body = fmt.Sprintf(`{"signed_execution_payload_bid":{"message":{`+ - `"block_hash":%q,"blob_kzg_commitments":%s}}}`, elBlockHash, commitsJSON) + body = fmt.Sprintf(`{"signed_execution_payload_bid":{"message":{"block_hash":%q%s}}}`, + elBlockHash, commitmentsField) } else { - body = fmt.Sprintf(`{"execution_payload":{"block_hash":%q},"blob_kzg_commitments":%s}`, - elBlockHash, commitsJSON) + body = fmt.Sprintf(`{"execution_payload":{"block_hash":%q}%s}`, elBlockHash, commitmentsField) } return fmt.Sprintf(`{"version":%q,"data":{"message":{"slot":%q,"body":%s}}}`, version, slot, body) } @@ -69,7 +80,7 @@ func TestParseBeaconBlockVersions(t *testing.T) { // the signed bid. {"gloas", beaconBlockVersionGloas, true, []string{testCommit0, testCommit1}}, {"gloas without blobs", beaconBlockVersionGloas, true, []string{}}, - {"deneb without blobs", beaconBlockVersionDeneb, false, nil}, + {"deneb without blobs", beaconBlockVersionDeneb, false, []string{}}, {"version case insensitive", "Gloas", true, []string{testCommit0}}, } for _, tc := range cases { @@ -135,6 +146,30 @@ func TestParseBeaconBlockErrors(t *testing.T) { `{"blob_kzg_commitments":[` + fmt.Sprintf("%q", testCommit0) + `]}}}}}}`, want: errUpstreamError, }, + // blob_kzg_commitments is required by the beacon API schema: a block + // with no blobs serves an explicit empty array. A missing or null + // field is a malformed upstream response, never "no blobs" (which + // would surface as a misleading 404). + { + name: "deneb with null blob_kzg_commitments", + body: beaconBlockJSON("deneb", "1", testBlockHash, nil, false), + want: errUpstreamError, + }, + { + name: "deneb without blob_kzg_commitments", + body: beaconBlockJSON("deneb", "1", testBlockHash, omitCommitments, false), + want: errUpstreamError, + }, + { + name: "gloas with null blob_kzg_commitments", + body: beaconBlockJSON("gloas", "1", testBlockHash, nil, true), + want: errUpstreamError, + }, + { + name: "gloas without blob_kzg_commitments", + body: beaconBlockJSON("gloas", "1", testBlockHash, omitCommitments, true), + want: errUpstreamError, + }, { name: "execution payload without block hash", body: `{"version":"deneb","data":{"message":{"slot":"1","body":{"blob_kzg_commitments":null}}}}`, diff --git a/ethstorage/eth/beacon_client.go b/ethstorage/eth/beacon_client.go index 458e6c0b..cd457327 100644 --- a/ethstorage/eth/beacon_client.go +++ b/ethstorage/eth/beacon_client.go @@ -10,7 +10,6 @@ import ( "net/http" "net/url" "strconv" - "time" "github.com/ethereum/go-ethereum/common" "github.com/ethstorage/go-ethstorage/ethstorage/blobs" @@ -20,7 +19,6 @@ type BeaconClient struct { beaconURL string genesisSlotTime uint64 slotTime uint64 - httpClient *http.Client } type Blob struct { @@ -37,7 +35,6 @@ func NewBeaconClient(url string, slotTime uint64) (*BeaconClient, error) { beaconURL: url, genesisSlotTime: genesisSlotTime, slotTime: slotTime, - httpClient: &http.Client{Timeout: 30 * time.Second}, } return res, nil } @@ -78,7 +75,7 @@ func (c *BeaconClient) DownloadBlobs(timestamp uint64, hashes []common.Hash) (ma if err != nil { return nil, err } - resp, err := c.httpClient.Get(blobsUrl) + resp, err := http.Get(blobsUrl) if err != nil { return nil, fmt.Errorf("failed to query beacon blobs with url %s: %w", blobsUrl, err) } @@ -128,13 +125,3 @@ func (c *BeaconClient) QueryUrlForV1BeaconBlobs(timestamp uint64, hashes []commo } return blobsURL + "?" + q.Encode(), nil } - -// HTTPClient returns the HTTP client used to query the beacon node, so that -// callers can issue requests with their own context (cancellation and -// timeouts) instead of fire-and-forget http.Get. -func (c *BeaconClient) HTTPClient() *http.Client { - if c.httpClient != nil { - return c.httpClient - } - return http.DefaultClient -} From 17ad3797db4e0b055270e63df4b0ac27507de4f8 Mon Sep 17 00:00:00 2001 From: syntrust Date: Mon, 12 Oct 2026 11:57:06 +0800 Subject: [PATCH 3/4] archiver: document beacon response size cap --- ethstorage/archiver/api.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/ethstorage/archiver/api.go b/ethstorage/archiver/api.go index d23bdd45..227c0950 100644 --- a/ethstorage/archiver/api.go +++ b/ethstorage/archiver/api.go @@ -30,8 +30,10 @@ import ( const ( // beaconBlockFetchTimeout bounds a single upstream beacon node request. beaconBlockFetchTimeout = 30 * time.Second - // maxBeaconBlockResponseSize caps how much of a beacon block response we - // read into memory (a full mainnet block is well under this). + // maxBeaconBlockResponseSize caps memory use while decoding a beacon block. + // The consensus P2P payload limit is 10 MiB; JSON hex-encodes execution + // transactions, so 64 MiB leaves headroom for encoding and envelope overhead. + // This is a defensive limit, not a Beacon API response-size guarantee. maxBeaconBlockResponseSize = 64 << 20 // beaconBlockVersionDeneb .. beaconBlockVersionGloas are the consensus // versions whose blocks the archiver knows how to decode. From 1a2603302553a1c80d0d6ffca54435639785f904 Mon Sep 17 00:00:00 2001 From: syntrust Date: Wed, 28 Oct 2026 15:47:31 +0800 Subject: [PATCH 4/4] archiver: preserve duplicate blob commitment positions --- ethstorage/archiver/api.go | 48 ++++++++----- ethstorage/archiver/api_e2e_test.go | 107 ++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 17 deletions(-) diff --git a/ethstorage/archiver/api.go b/ethstorage/archiver/api.go index 227c0950..ee7a8579 100644 --- a/ethstorage/archiver/api.go +++ b/ethstorage/archiver/api.go @@ -107,14 +107,14 @@ func (a *API) queryBlobs(ctx context.Context, id string, hashes []common.Hash) ( return nil, hErr } - // hashToIndex is used to determine correct blob index - hashToIndex := make(map[common.Hash]int) - for _, c := range block.kzgCommitsAll { + // A blob hash can occur at multiple positions in the block. + hashToIndices := make(map[common.Hash][]int) + for i, c := range block.kzgCommitsAll { var commit kzg4844.Commitment copy(commit[:], common.FromHex(c)) vh := common.Hash(kzg4844.CalcBlobHashV1(sha256.New(), &commit)) if hashIncluded(vh, hashes) { - hashToIndex[vh] = len(hashToIndex) + hashToIndices[vh] = append(hashToIndices[vh], i) } } @@ -124,19 +124,23 @@ func (a *API) queryBlobs(ctx context.Context, id string, hashes []common.Hash) ( return nil, hErr } - blobsData := make([]string, len(hashToIndex)) + blobsData := make([]string, len(block.kzgCommitsAll)) for i, event := range events { blobHash := event.Topics[3] a.lg.Info("Parsing event", "blobHash", blobHash, "event", fmt.Sprintf("%d of %d", i, len(events))) // parse event to get kv_index with queried index - if index, ok := hashToIndex[blobHash]; ok { + if indices, ok := hashToIndices[blobHash]; ok { kvIndex := big.NewInt(0).SetBytes(event.Topics[1][:]).Uint64() - a.lg.Info("Blobhash matched", "blobhash", blobHash, "index", index, "kvIndex", kvIndex) + a.lg.Info("Blobhash matched", "blobhash", blobHash, "indices", indices, "kvIndex", kvIndex) blobData, hErr := a.readBlob(kvIndex, blobHash) if hErr != nil { return nil, hErr } - blobsData[index] = "0x" + common.Bytes2Hex(blobData) + encodedBlob := "0x" + common.Bytes2Hex(blobData) + for _, index := range indices { + blobsData[index] = encodedBlob + } + delete(hashToIndices, blobHash) } } // remove empty entries (not found) @@ -166,12 +170,12 @@ func (a *API) queryBlobSidecars(ctx context.Context, id string, indices []uint64 } } - // hashToIndex is used to determine correct blob index - hashToIndex := make(map[common.Hash]Index) + // Preserve every requested position, including repeated commitments. + hashToIndices := make(map[common.Hash][]Index) for i, c := range kzgCommitsAll { if indexIncluded(uint64(i), indices) { bh := gkzg.KZGToVersionedHash(gkzg.KZGCommitment(common.FromHex(c))) - hashToIndex[common.Hash(bh)] = Index(i) + hashToIndices[common.Hash(bh)] = append(hashToIndices[common.Hash(bh)], Index(i)) } } @@ -180,21 +184,31 @@ func (a *API) queryBlobSidecars(ctx context.Context, id string, indices []uint64 if hErr != nil { return nil, hErr } - var res BlobSidecars + sidecars := make([]*BlobSidecar, len(kzgCommitsAll)) for i, event := range events { blobHash := event.Topics[3] a.lg.Info("Parsing event", "blobHash", blobHash, "event", fmt.Sprintf("%d of %d", i, len(events))) // parse event to get kv_index with queried index - if index, ok := hashToIndex[blobHash]; ok { + if blobIndices, ok := hashToIndices[blobHash]; ok { kvIndex := big.NewInt(0).SetBytes(event.Topics[1][:]).Uint64() - a.lg.Info("Blobhash matched", "blobhash", blobHash, "index", index, "kvIndex", kvIndex) - sidecar, hErr := a.buildSidecar(kvIndex, common.FromHex(kzgCommitsAll[index]), blobHash) + a.lg.Info("Blobhash matched", "blobhash", blobHash, "indices", blobIndices, "kvIndex", kvIndex) + sidecar, hErr := a.buildSidecar(kvIndex, common.FromHex(kzgCommitsAll[blobIndices[0]]), blobHash) if hErr != nil { a.lg.Error("Failed to build sidecar", "err", hErr) return nil, hErr } - sidecar.Index = index - a.lg.Info("Sidecar built", "index", index, "sidecar", sidecar) + for _, index := range blobIndices { + indexedSidecar := *sidecar + indexedSidecar.Index = index + a.lg.Info("Sidecar built", "index", index, "sidecar", &indexedSidecar) + sidecars[index] = &indexedSidecar + } + delete(hashToIndices, blobHash) + } + } + var res BlobSidecars + for _, sidecar := range sidecars { + if sidecar != nil { res.Data = append(res.Data, sidecar) } } diff --git a/ethstorage/archiver/api_e2e_test.go b/ethstorage/archiver/api_e2e_test.go index 0a5a573d..33a15fe4 100644 --- a/ethstorage/archiver/api_e2e_test.go +++ b/ethstorage/archiver/api_e2e_test.go @@ -307,6 +307,113 @@ func TestQueryBlobSidecarsGloas(t *testing.T) { assertGetLogsByHash(t, h.getLogsReq) } +func TestQueryBlobsDuplicateCommitments(t *testing.T) { + kvIndexes := []uint64{2, 7} + raw := make([][]byte, 2) + commits := make([]kzg4844.Commitment, 2) + hashes := make([]common.Hash, 2) + for i, kv := range kvIndexes { + raw[i], commits[i], hashes[i] = makeBlob(t, kv) + } + storageMgr := newStorageManager(t, kvIndexes, raw, hashes) + // Repeated PutBlob events must not change the number of returned blobs. + elLogs := []types.Log{ + putBlobLog(kvIndexes[1], hashes[1], 0), + putBlobLog(kvIndexes[0], hashes[0], 1), + putBlobLog(kvIndexes[0], hashes[0], 2), + } + + for _, tc := range []struct { + name string + order []int + hashes []common.Hash + want []int + }{ + {"ABA_all", []int{0, 1, 0}, nil, []int{0, 1, 0}}, + {"AAB_all", []int{0, 0, 1}, nil, []int{0, 0, 1}}, + {"ABA_filter_repeated", []int{0, 1, 0}, []common.Hash{hashes[0]}, []int{0, 0}}, + {"AAB_filter_unique", []int{0, 0, 1}, []common.Hash{hashes[1]}, []int{1}}, + {"AAB_filter_reversed", []int{0, 0, 1}, []common.Hash{hashes[1], hashes[0]}, []int{0, 0, 1}}, + } { + t.Run(tc.name, func(t *testing.T) { + commitHex := make([]string, len(tc.order)) + for i, blob := range tc.order { + commitHex[i] = "0x" + common.Bytes2Hex(commits[blob][:]) + } + h := newE2EAPI(t, storageMgr, beaconBlockVersionGloas, commitHex, elLogs, nil) + res, hErr := h.api.queryBlobs(context.Background(), e2eSlot, tc.hashes) + if hErr != nil { + t.Fatalf("queryBlobs failed: %d %s", hErr.Code, hErr.Message) + } + if len(res.Data) != len(tc.want) { + t.Fatalf("got %d blobs, want %d", len(res.Data), len(tc.want)) + } + for i, blob := range tc.want { + if res.Data[i] != "0x"+common.Bytes2Hex(raw[blob]) { + t.Errorf("blob %d does not match blob %d in commitment order", i, blob) + } + } + }) + } +} + +func TestQueryBlobSidecarsDuplicateCommitments(t *testing.T) { + kvIndexes := []uint64{2, 7} + raw := make([][]byte, 2) + commits := make([]kzg4844.Commitment, 2) + hashes := make([]common.Hash, 2) + for i, kv := range kvIndexes { + raw[i], commits[i], hashes[i] = makeBlob(t, kv) + } + storageMgr := newStorageManager(t, kvIndexes, raw, hashes) + elLogs := []types.Log{ + putBlobLog(kvIndexes[1], hashes[1], 0), + putBlobLog(kvIndexes[0], hashes[0], 1), + putBlobLog(kvIndexes[0], hashes[0], 2), + } + + for _, tc := range []struct { + name string + order []int + indices []uint64 + want []Index + }{ + {"ABA_all", []int{0, 1, 0}, nil, []Index{0, 1, 2}}, + {"AAB_all", []int{0, 0, 1}, nil, []Index{0, 1, 2}}, + {"ABA_filter_both_repeated", []int{0, 1, 0}, []uint64{2, 0}, []Index{0, 2}}, + {"AAB_filter_one_repeated", []int{0, 0, 1}, []uint64{1}, []Index{1}}, + } { + t.Run(tc.name, func(t *testing.T) { + commitHex := make([]string, len(tc.order)) + for i, blob := range tc.order { + commitHex[i] = "0x" + common.Bytes2Hex(commits[blob][:]) + } + h := newE2EAPI(t, storageMgr, beaconBlockVersionGloas, commitHex, elLogs, nil) + res, hErr := h.api.queryBlobSidecars(context.Background(), e2eSlot, tc.indices) + if hErr != nil { + t.Fatalf("queryBlobSidecars failed: %d %s", hErr.Code, hErr.Message) + } + if len(res.Data) != len(tc.want) { + t.Fatalf("got %d sidecars, want %d", len(res.Data), len(tc.want)) + } + for i, index := range tc.want { + sidecar := res.Data[i] + if sidecar.Index != index { + t.Errorf("sidecar %d index = %d, want %d", i, sidecar.Index, index) + } + blob := tc.order[index] + if !equalBytes(sidecar.Blob[:], raw[blob]) || !equalBytes(sidecar.KZGCommitment[:], commits[blob][:]) { + t.Errorf("sidecar %d does not match the blob and commitment at index %d", i, index) + } + sidecarBlob := kzg4844.Blob(sidecar.Blob) + if err := kzg4844.VerifyBlobProof(&sidecarBlob, kzg4844.Commitment(sidecar.KZGCommitment), kzg4844.Proof(sidecar.KZGProof)); err != nil { + t.Errorf("sidecar %d has an invalid proof: %v", i, err) + } + } + }) + } +} + // TestQueryBlobsStorageMissing covers a block that exists on both the beacon // chain and L1, but whose blob is not stored locally. func TestQueryBlobsStorageMissing(t *testing.T) {