forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 14:46:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core, cmd: change custodyBitmap pointer to value

Browse source
This commit is contained in:
healthykim 2026-06-15 22:38:50 +02:00
parent 1316b2ee79
commit 2ed436cbc0
11 changed files with 121 additions and 123 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
cmd/devp2p/internal/ethtest/suite.go
View file

@ -1074,7 +1074,7 @@ func (s *Suite) TestBlobViolations(t *utesting.T) {
Types: []byte{types.BlobTxType, types.BlobTxType}, Types: []byte{types.BlobTxType, types.BlobTxType},
Sizes: []uint32{uint32(t1[0].Size()), uint32(t1[1].Size() + 10)}, Sizes: []uint32{uint32(t1[0].Size()), uint32(t1[1].Size() + 10)},
Hashes: []common.Hash{t1[0].Hash(), t1[1].Hash()}, Hashes: []common.Hash{t1[0].Hash(), t1[1].Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
}, },
resp: eth.PooledTransactionsResponse(t1), resp: eth.PooledTransactionsResponse(t1),
}, },
@ -1084,7 +1084,7 @@ func (s *Suite) TestBlobViolations(t *utesting.T) {
Types: []byte{types.DynamicFeeTxType, types.BlobTxType}, Types: []byte{types.DynamicFeeTxType, types.BlobTxType},
Sizes: []uint32{uint32(t2[0].Size()), uint32(t2[1].Size())}, Sizes: []uint32{uint32(t2[0].Size()), uint32(t2[1].Size())},
Hashes: []common.Hash{t2[0].Hash(), t2[1].Hash()}, Hashes: []common.Hash{t2[0].Hash(), t2[1].Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
}, },
resp: eth.PooledTransactionsResponse(t2), resp: eth.PooledTransactionsResponse(t2),
}, },
@ -1226,7 +1226,7 @@ func (s *Suite) testBadBlobTx(t *utesting.T, tx *types.Transaction, badTx *types
Types: []byte{types.BlobTxType}, Types: []byte{types.BlobTxType},
Sizes: []uint32{uint32(badTx.Size())}, Sizes: []uint32{uint32(badTx.Size())},
Hashes: []common.Hash{badTx.Hash()}, Hashes: []common.Hash{badTx.Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
} }
if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil { if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
@ -1278,7 +1278,7 @@ func (s *Suite) testBadBlobTx(t *utesting.T, tx *types.Transaction, badTx *types
Types: []byte{types.BlobTxType}, Types: []byte{types.BlobTxType},
Sizes: []uint32{uint32(tx.Size())}, Sizes: []uint32{uint32(tx.Size())},
Hashes: []common.Hash{tx.Hash()}, Hashes: []common.Hash{tx.Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
} }
if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil { if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
@ -1366,7 +1366,7 @@ partial fetch GetCells should never arrive. Any GetCells that does arrive must b
Types: txTypes, Types: txTypes,
Sizes: sizes, Sizes: sizes,
Hashes: hashes, Hashes: hashes,
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
} }
if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil { if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
t.Fatalf("announce failed: %v", err) t.Fatalf("announce failed: %v", err)
@ -1485,7 +1485,7 @@ and that providing valid cells causes the tx to enter the pool.`)
Types: []byte{types.BlobTxType}, Types: []byte{types.BlobTxType},
Sizes: []uint32{uint32(tx.Size())}, Sizes: []uint32{uint32(tx.Size())},
Hashes: []common.Hash{tx.Hash()}, Hashes: []common.Hash{tx.Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
} }
if err := conn1.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil { if err := conn1.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
t.Fatalf("conn1 announce failed: %v", err) t.Fatalf("conn1 announce failed: %v", err)
@ -1568,7 +1568,7 @@ while the other peer is not.`)
Types: []byte{types.BlobTxType}, Types: []byte{types.BlobTxType},
Sizes: []uint32{uint32(tx.Size())}, Sizes: []uint32{uint32(tx.Size())},
Hashes: []common.Hash{tx.Hash()}, Hashes: []common.Hash{tx.Hash()},
Mask: *types.CustodyBitmapAll, Mask: types.CustodyBitmapAll,
} }
if err := conn1.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil { if err := conn1.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
t.Fatalf("conn1 announce failed: %v", err) t.Fatalf("conn1 announce failed: %v", err)

2
core/txpool/blobpool/blobpool.go
View file

@ -225,7 +225,7 @@ func newBlobTxForPool(tx *types.Transaction) (*BlobTxForPool, error) {
Commitments: sc.Commitments, Commitments: sc.Commitments,
Proofs: sc.Proofs, Proofs: sc.Proofs,
Cells: cells, Cells: cells,
Custody: *types.CustodyBitmapAll, Custody: types.CustodyBitmapAll,
} }
return &BlobTxForPool{ return &BlobTxForPool{
Tx: tx.WithoutBlobTxSidecar(), Tx: tx.WithoutBlobTxSidecar(),

2
core/txpool/blobpool/blobpool_test.go
View file

@ -2292,7 +2292,7 @@ func TestGetCells(t *testing.T) {
{ {
name: "Get cells with all indices", name: "Get cells with all indices",
hash: tx1.Hash(), hash: tx1.Hash(),
mask: *types.CustodyBitmapAll, mask: types.CustodyBitmapAll,
expectedLen: 384, expectedLen: 384,
shouldFail: false, shouldFail: false,
}, },

10
core/txpool/blobpool/buffer.go
View file

@ -59,7 +59,7 @@ type txEntry struct {
type cellEntry struct { type cellEntry struct {
deliveries map[string]*PeerDelivery deliveries map[string]*PeerDelivery
custody *types.CustodyBitmap custody types.CustodyBitmap
added time.Time added time.Time
} }
@ -145,7 +145,7 @@ func (b *BlobBuffer) AddTx(txs []*types.Transaction, peer string) []error {
// AddCells buffers per-peer cell deliveries from the blob fetcher. // AddCells buffers per-peer cell deliveries from the blob fetcher.
// If the transaction is already buffered, verification and pool insertion are attempted. // If the transaction is already buffered, verification and pool insertion are attempted.
func (b *BlobBuffer) AddCells(hash common.Hash, deliveries map[string]*PeerDelivery, custody *types.CustodyBitmap) { func (b *BlobBuffer) AddCells(hash common.Hash, deliveries map[string]*PeerDelivery, custody types.CustodyBitmap) {
b.mu.Lock() b.mu.Lock()
defer b.mu.Unlock() defer b.mu.Unlock()
defer b.updateMetrics()() defer b.updateMetrics()()
@ -182,7 +182,7 @@ func (b *BlobBuffer) add(hash common.Hash, tx *types.Transaction, cells *cellEnt
Commitments: sidecar.Commitments, Commitments: sidecar.Commitments,
Proofs: sidecar.Proofs, Proofs: sidecar.Proofs,
Cells: sorted, Cells: sorted,
Custody: *custody, Custody: custody,
} }
pooledTx := &BlobTxForPool{ pooledTx := &BlobTxForPool{
Tx: tx.WithoutBlobTxSidecar(), Tx: tx.WithoutBlobTxSidecar(),
@ -277,7 +277,7 @@ func (b *BlobBuffer) verifyCells(entry *cellEntry, sidecar *types.BlobTxSidecar)
// peer A: cells = [blob0_cell5, blob0_cell3, blob1_cell5, blob1_cell3] // peer A: cells = [blob0_cell5, blob0_cell3, blob1_cell5, blob1_cell3]
// peer B: cells = [blob0_cell1, blob0_cell7, blob1_cell1, blob1_cell7] // peer B: cells = [blob0_cell1, blob0_cell7, blob1_cell1, blob1_cell7]
// -> [blob0_cell1, blob0_cell3, blob0_cell5, blob0_cell7, blob1_cell1, blob1_cell3, blob1_cell5, blob1_cell7] // -> [blob0_cell1, blob0_cell3, blob0_cell5, blob0_cell7, blob1_cell1, blob1_cell3, blob1_cell5, blob1_cell7]
func sortCells(entry *cellEntry, blobCount int) ([]kzg4844.Cell, *types.CustodyBitmap) { func sortCells(entry *cellEntry, blobCount int) ([]kzg4844.Cell, types.CustodyBitmap) {
// indices per delivery // indices per delivery
var indices []uint64 var indices []uint64
@ -309,5 +309,5 @@ func sortCells(entry *cellEntry, blobCount int) ([]kzg4844.Cell, *types.CustodyB
} }
custody := types.NewCustodyBitmap(indices) custody := types.NewCustodyBitmap(indices)
return res, &custody return res, custody
} }

10
core/txpool/blobpool/buffer_test.go
View file

@ -59,7 +59,7 @@ func TestSortCells(t *testing.T) {
"peerA": peerA, "peerA": peerA,
"peerB": peerB, "peerB": peerB,
}, },
custody: &custody, custody: custody,
} }
sorted, resultCustody := sortCells(entry, blobCount) sorted, resultCustody := sortCells(entry, blobCount)
@ -106,7 +106,7 @@ func TestAddTxThenCells(t *testing.T) {
delivery := makePeerDelivery(t, 0, blobCount, dataIndices) delivery := makePeerDelivery(t, 0, blobCount, dataIndices)
custody := types.NewCustodyBitmap(dataIndices) custody := types.NewCustodyBitmap(dataIndices)
buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, &custody) buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, custody)
if buf.HasTx(hash) || buf.HasCells(hash) { if buf.HasTx(hash) || buf.HasCells(hash) {
t.Fatal("buffer should be empty after add") t.Fatal("buffer should be empty after add")
} }
@ -127,7 +127,7 @@ func TestAddCellsThenTx(t *testing.T) {
delivery := makePeerDelivery(t, 0, blobCount, dataIndices) delivery := makePeerDelivery(t, 0, blobCount, dataIndices)
custody := types.NewCustodyBitmap(dataIndices) custody := types.NewCustodyBitmap(dataIndices)
buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, &custody) buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, custody)
if !buf.HasCells(hash) { if !buf.HasCells(hash) {
t.Fatal("cells should be buffered") t.Fatal("cells should be buffered")
} }
@ -160,7 +160,7 @@ func TestMultiPeerDelivery(t *testing.T) {
buf.AddCells(hash, map[string]*PeerDelivery{ buf.AddCells(hash, map[string]*PeerDelivery{
"peerB": deliveryA, "peerB": deliveryA,
"peerC": deliveryB, "peerC": deliveryB,
}, &custody) }, custody)
if buf.HasTx(hash) || buf.HasCells(hash) { if buf.HasTx(hash) || buf.HasCells(hash) {
t.Fatal("buffer should be empty after add") t.Fatal("buffer should be empty after add")
} }
@ -195,7 +195,7 @@ func TestBadCell(t *testing.T) {
buf.AddCells(hash, map[string]*PeerDelivery{ buf.AddCells(hash, map[string]*PeerDelivery{
"peerB": goodDelivery, "peerB": goodDelivery,
"peerC": badDelivery, "peerC": badDelivery,
}, &custody) }, custody)
if len(dropped) != 1 || dropped[0] != "peerC" { if len(dropped) != 1 || dropped[0] != "peerC" {
t.Fatalf("only peerC should have been dropped, got: %v", dropped) t.Fatalf("only peerC should have been dropped, got: %v", dropped)

20
core/types/custody_bitmap.go
View file

@ -47,20 +47,20 @@ func (b *CustodyBitmap) UnmarshalText(input []byte) error {
} }
var ( var (
CustodyBitmapAll = func() *CustodyBitmap { CustodyBitmapAll = func() CustodyBitmap {
var result CustodyBitmap var result CustodyBitmap
for i := range result { for i := range result {
result[i] = 0xFF result[i] = 0xFF
} }
return &result return result
}() }()
CustodyBitmapData = func() *CustodyBitmap { CustodyBitmapData = func() CustodyBitmap {
var result CustodyBitmap var result CustodyBitmap
for i := 0; i < kzg4844.DataPerBlob/8; i++ { for i := 0; i < kzg4844.DataPerBlob/8; i++ {
result[i] = 0xFF result[i] = 0xFF
} }
return &result return result
}() }()
) )
@ -125,28 +125,28 @@ func (b CustodyBitmap) Indices() []uint64 {
} }
// Difference returns b AND NOT set (bits in b but not in set). // Difference returns b AND NOT set (bits in b but not in set).
func (b CustodyBitmap) Difference(set *CustodyBitmap) *CustodyBitmap { func (b CustodyBitmap) Difference(set CustodyBitmap) CustodyBitmap {
var out CustodyBitmap var out CustodyBitmap
for i := range b { for i := range b {
out[i] = b[i] &^ set[i] out[i] = b[i] &^ set[i]
} }
return &out return out
} }
// Intersection returns b AND set. // Intersection returns b AND set.
func (b CustodyBitmap) Intersection(set *CustodyBitmap) *CustodyBitmap { func (b CustodyBitmap) Intersection(set CustodyBitmap) CustodyBitmap {
var out CustodyBitmap var out CustodyBitmap
for i := range b { for i := range b {
out[i] = b[i] & set[i] out[i] = b[i] & set[i]
} }
return &out return out
} }
// Union returns b OR set. // Union returns b OR set.
func (b CustodyBitmap) Union(set *CustodyBitmap) *CustodyBitmap { func (b CustodyBitmap) Union(set CustodyBitmap) CustodyBitmap {
var out CustodyBitmap var out CustodyBitmap
for i := range b { for i := range b {
out[i] = b[i] | set[i] out[i] = b[i] | set[i]
} }
return &out return out
} }

2
eth/backend.go
View file

@ -356,7 +356,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
Sync: config.SyncMode, Sync: config.SyncMode,
BloomCache: uint64(cacheLimit), BloomCache: uint64(cacheLimit),
RequiredBlocks: config.RequiredBlocks, RequiredBlocks: config.RequiredBlocks,
Custody: *types.CustodyBitmapAll, Custody: types.CustodyBitmapAll,
SnapV2: config.SnapV2, SnapV2: config.SnapV2,
}); err != nil { }); err != nil {
return nil, err return nil, err

54
eth/fetcher/blob_fetcher.go
View file

@ -63,21 +63,21 @@ type blobTxAnnounce struct {
} }
type cellRequest struct { type cellRequest struct {
txs []common.Hash // Transactions that have been requested for their cells txs []common.Hash // Transactions that have been requested for their cells
cells *types.CustodyBitmap // Requested cell indices cells types.CustodyBitmap // Requested cell indices
time mclock.AbsTime // Timestamp when the request was made time mclock.AbsTime // Timestamp when the request was made
} }
type payloadDelivery struct { type payloadDelivery struct {
origin string // Peer from which the payloads were delivered origin string // Peer from which the payloads were delivered
txs []common.Hash // Hashes of transactions that were delivered txs []common.Hash // Hashes of transactions that were delivered
cells [][]kzg4844.Cell cells [][]kzg4844.Cell
cellBitmap *types.CustodyBitmap cellBitmap types.CustodyBitmap
} }
type cellWithSeq struct { type cellWithSeq struct {
seq uint64 seq uint64
cells *types.CustodyBitmap cells types.CustodyBitmap
} }
// PeerCellDelivery holds cells delivered by a single peer. // PeerCellDelivery holds cells delivered by a single peer.
@ -87,7 +87,7 @@ type PeerCellDelivery struct {
} }
type fetchStatus struct { type fetchStatus struct {
fetching *types.CustodyBitmap // To avoid fetching cells which had already been fetched / currently being fetched fetching types.CustodyBitmap // To avoid fetching cells which had already been fetched / currently being fetched
fetched []uint64 // Custody indices that have been fetched (per-blob, same for all blobs) fetched []uint64 // Custody indices that have been fetched (per-blob, same for all blobs)
deliveries map[string]*PeerCellDelivery // Per-peer cell deliveries deliveries map[string]*PeerCellDelivery // Per-peer cell deliveries
blobCount int // Number of blobs in this tx (set on first delivery) blobCount int // Number of blobs in this tx (set on first delivery)
@ -95,8 +95,8 @@ type fetchStatus struct {
type BlobFetcherFunctions struct { type BlobFetcherFunctions struct {
HasPayload func(common.Hash) bool HasPayload func(common.Hash) bool
AddCells func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) AddCells func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap)
FetchPayloads func(string, []common.Hash, *types.CustodyBitmap) error FetchPayloads func(string, []common.Hash, types.CustodyBitmap) error
DropPeer func(string) DropPeer func(string)
} }
@ -116,7 +116,7 @@ type BlobFetcher struct {
drop chan *txDrop drop chan *txDrop
custodyCh chan types.CustodyBitmap custodyCh chan types.CustodyBitmap
quit chan struct{} quit chan struct{}
custody *types.CustodyBitmap custody types.CustodyBitmap
txSeq uint64 // To make transactions fetched in arrival order txSeq uint64 // To make transactions fetched in arrival order
@ -135,9 +135,9 @@ type BlobFetcher struct {
// Buffer 2 // Buffer 2
// Stage 3: Transactions whose payloads/cells are currently being fetched (full fetch + partial fetch) // Stage 3: Transactions whose payloads/cells are currently being fetched (full fetch + partial fetch)
fetches map[common.Hash]*fetchStatus // Hash -> Bitmap, in-flight transaction cells fetches map[common.Hash]*fetchStatus // Hash -> Bitmap, in-flight transaction cells
requests map[string][]*cellRequest // In-flight transaction retrievals requests map[string][]*cellRequest // In-flight transaction retrievals
alternates map[common.Hash]map[string]*types.CustodyBitmap // In-flight transaction alternate origins (in case the peer is dropped) alternates map[common.Hash]map[string]types.CustodyBitmap // In-flight transaction alternate origins (in case the peer is dropped)
fn BlobFetcherFunctions // callbacks fn BlobFetcherFunctions // callbacks
@ -156,7 +156,7 @@ type token struct {
last mclock.AbsTime last mclock.AbsTime
} }
func NewBlobFetcher(fn BlobFetcherFunctions, custody *types.CustodyBitmap, rand random) *BlobFetcher { func NewBlobFetcher(fn BlobFetcherFunctions, custody types.CustodyBitmap, rand random) *BlobFetcher {
return &BlobFetcher{ return &BlobFetcher{
notify: make(chan *blobTxAnnounce), notify: make(chan *blobTxAnnounce),
cleanup: make(chan *payloadDelivery), cleanup: make(chan *payloadDelivery),
@ -171,7 +171,7 @@ func NewBlobFetcher(fn BlobFetcherFunctions, custody *types.CustodyBitmap, rand
announces: make(map[string]map[common.Hash]*cellWithSeq), announces: make(map[string]map[common.Hash]*cellWithSeq),
fetches: make(map[common.Hash]*fetchStatus), fetches: make(map[common.Hash]*fetchStatus),
requests: make(map[string][]*cellRequest), requests: make(map[string][]*cellRequest),
alternates: make(map[common.Hash]map[string]*types.CustodyBitmap), alternates: make(map[common.Hash]map[string]types.CustodyBitmap),
peerTokens: make(map[string]*token), peerTokens: make(map[string]*token),
fn: fn, fn: fn,
custody: custody, custody: custody,
@ -207,7 +207,7 @@ func (f *BlobFetcher) Enqueue(peer string, hashes []common.Hash, cells [][]kzg48
blobReplyInMeter.Mark(int64(len(hashes))) blobReplyInMeter.Mark(int64(len(hashes)))
select { select {
case f.cleanup <- &payloadDelivery{origin: peer, txs: hashes, cells: cells, cellBitmap: &cellBitmap}: case f.cleanup <- &payloadDelivery{origin: peer, txs: hashes, cells: cells, cellBitmap: cellBitmap}:
case <-f.quit: case <-f.quit:
return errTerminated return errTerminated
} }
@ -306,7 +306,7 @@ func (f *BlobFetcher) loop() {
} }
if _, ok := f.full[hash]; ok { if _, ok := f.full[hash]; ok {
// 1) Decided to send full request of the tx // 1) Decided to send full request of the tx
if ann.cells != *types.CustodyBitmapAll { if ann.cells != types.CustodyBitmapAll {
continue continue
} }
if f.announces[ann.origin] == nil { if f.announces[ann.origin] == nil {
@ -322,7 +322,7 @@ func (f *BlobFetcher) loop() {
if _, ok := f.partial[hash]; ok { if _, ok := f.partial[hash]; ok {
// 2) Decided to send partial request of the tx // 2) Decided to send partial request of the tx
if f.waitlist[hash] != nil { if f.waitlist[hash] != nil {
if ann.cells != *types.CustodyBitmapAll { if ann.cells != types.CustodyBitmapAll {
// Availability check is only meaningful with full availability announcements // Availability check is only meaningful with full availability announcements
continue continue
} }
@ -473,7 +473,7 @@ func (f *BlobFetcher) loop() {
for _, hash := range delivery.txs { for _, hash := range delivery.txs {
// Find the request // Find the request
for i, req := range f.requests[delivery.origin] { for i, req := range f.requests[delivery.origin] {
if slices.Contains(req.txs, hash) && *req.cells == *delivery.cellBitmap { if slices.Contains(req.txs, hash) && req.cells == delivery.cellBitmap {
request = req request = req
requestId = i requestId = i
break break
@ -537,7 +537,7 @@ func (f *BlobFetcher) loop() {
blobFetcherFetchTime.Update(int64(time.Duration(f.clock.Now() - request.time))) blobFetcherFetchTime.Update(int64(time.Duration(f.clock.Now() - request.time)))
status := f.fetches[hash] status := f.fetches[hash]
collectedCustody := types.NewCustodyBitmap(status.fetched) collectedCustody := types.NewCustodyBitmap(status.fetched)
f.fn.AddCells(hash, status.deliveries, &collectedCustody) f.fn.AddCells(hash, status.deliveries, collectedCustody)
for peer, txset := range f.announces { for peer, txset := range f.announces {
delete(txset, hash) delete(txset, hash)
@ -640,7 +640,7 @@ func (f *BlobFetcher) loop() {
} }
case cells := <-f.custodyCh: case cells := <-f.custodyCh:
f.custody = &cells f.custody = cells
case <-f.quit: case <-f.quit:
return return
@ -755,10 +755,10 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
} }
var ( var (
hashes = make([]common.Hash, 0, maxTxRetrievals) hashes = make([]common.Hash, 0, maxTxRetrievals)
custodies = make([]*types.CustodyBitmap, 0, maxTxRetrievals) custodies = make([]types.CustodyBitmap, 0, maxTxRetrievals)
) )
f.forEachAnnounce(f.announces[peer], func(hash common.Hash, cells *types.CustodyBitmap) bool { f.forEachAnnounce(f.announces[peer], func(hash common.Hash, cells types.CustodyBitmap) bool {
var unfetched *types.CustodyBitmap var unfetched types.CustodyBitmap
if f.fetches[hash] == nil { if f.fetches[hash] == nil {
// tx is not being fetched // tx is not being fetched
@ -786,7 +786,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
// Mark alternatives // Mark alternatives
if f.alternates[hash] == nil { if f.alternates[hash] == nil {
f.alternates[hash] = map[string]*types.CustodyBitmap{ f.alternates[hash] = map[string]types.CustodyBitmap{
peer: cells, peer: cells,
} }
} else { } else {
@ -801,7 +801,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
requestByCustody := make(map[types.CustodyBitmap]*cellRequest) requestByCustody := make(map[types.CustodyBitmap]*cellRequest)
for i, hash := range hashes { for i, hash := range hashes {
key := *custodies[i] key := custodies[i]
if _, ok := requestByCustody[key]; !ok { if _, ok := requestByCustody[key]; !ok {
requestByCustody[key] = &cellRequest{ requestByCustody[key] = &cellRequest{
txs: []common.Hash{}, txs: []common.Hash{},
@ -839,10 +839,10 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
// the do function for each until it returns false. We enforce an arrival // the do function for each until it returns false. We enforce an arrival
// ordering to minimize the chances of transaction nonce-gaps, which result in // ordering to minimize the chances of transaction nonce-gaps, which result in
// transactions being rejected by the txpool. // transactions being rejected by the txpool.
func (f *BlobFetcher) forEachAnnounce(announces map[common.Hash]*cellWithSeq, do func(hash common.Hash, cells *types.CustodyBitmap) bool) { func (f *BlobFetcher) forEachAnnounce(announces map[common.Hash]*cellWithSeq, do func(hash common.Hash, cells types.CustodyBitmap) bool) {
type announcement struct { type announcement struct {
hash common.Hash hash common.Hash
cells *types.CustodyBitmap cells types.CustodyBitmap
seq uint64 seq uint64
} }
// Process announcements by their arrival order // Process announcements by their arrival order

120
eth/fetcher/blob_fetcher_test.go
View file

@ -53,11 +53,11 @@ func makeTestCellSidecar(blobCount int) *types.BlobTxCellSidecar {
Cells: cells, Cells: cells,
Commitments: commitments, Commitments: commitments,
Proofs: proofs, Proofs: proofs,
Custody: *types.CustodyBitmapAll, Custody: types.CustodyBitmapAll,
} }
} }
func selectCells(cells []kzg4844.Cell, custody *types.CustodyBitmap) []kzg4844.Cell { func selectCells(cells []kzg4844.Cell, custody types.CustodyBitmap) []kzg4844.Cell {
custodyIndices := custody.Indices() custodyIndices := custody.Indices()
result := make([]kzg4844.Cell, 0) result := make([]kzg4844.Cell, 0)
@ -82,8 +82,8 @@ var (
custody = types.NewCustodyBitmap([]uint64{0, 1, 2, 3, 4, 5, 6, 7}) custody = types.NewCustodyBitmap([]uint64{0, 1, 2, 3, 4, 5, 6, 7})
fullCustody = *types.CustodyBitmapAll fullCustody = types.CustodyBitmapAll
halfCustody = *types.CustodyBitmapData halfCustody = types.CustodyBitmapData
frontCustody = types.NewCustodyBitmap([]uint64{0, 1, 2, 3, 8, 9, 10, 11}) frontCustody = types.NewCustodyBitmap([]uint64{0, 1, 2, 3, 8, 9, 10, 11})
backCustody = types.NewCustodyBitmap([]uint64{4, 5, 6, 7, 8, 9, 10, 11}) backCustody = types.NewCustodyBitmap([]uint64{4, 5, 6, 7, 8, 9, 10, 11})
differentCustody = types.NewCustodyBitmap([]uint64{8, 9, 10, 11, 12, 13, 14, 15}) differentCustody = types.NewCustodyBitmap([]uint64{8, 9, 10, 11, 12, 13, 14, 15})
@ -154,13 +154,13 @@ func TestBlobFetcherFullFetch(t *testing.T) {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, HasPayload: func(common.Hash) bool { return false },
AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 5}, // Force full requests (5 < fetchProbability) &mockRand{value: 5}, // Force full requests (5 < fetchProbability)
) )
}, },
@ -244,14 +244,14 @@ func TestBlobFetcherPartialFetch(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 60}, // Force partial requests (20 >= 15) &mockRand{value: 60}, // Force partial requests (20 >= 15)
) )
}, },
@ -293,7 +293,7 @@ func TestBlobFetcherPartialFetch(t *testing.T) {
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"A": {{hash: testBlobTxHashes[0], custody: custody}}, "A": {{hash: testBlobTxHashes[0], custody: custody}},
"B": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "B": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"C": {{hash: testBlobTxHashes[0], custody: custody}}, "C": {{hash: testBlobTxHashes[0], custody: custody}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
@ -306,7 +306,7 @@ func TestBlobFetcherPartialFetch(t *testing.T) {
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"A": {{hash: testBlobTxHashes[0], custody: custody}}, "A": {{hash: testBlobTxHashes[0], custody: custody}},
"B": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "B": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"C": {{hash: testBlobTxHashes[0], custody: custody}}, "C": {{hash: testBlobTxHashes[0], custody: custody}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
@ -319,7 +319,7 @@ func TestBlobFetcherPartialFetch(t *testing.T) {
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"A": {{hash: testBlobTxHashes[0], custody: custody}}, "A": {{hash: testBlobTxHashes[0], custody: custody}},
"B": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "B": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"C": {{hash: testBlobTxHashes[0], custody: custody}}, "C": {{hash: testBlobTxHashes[0], custody: custody}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
@ -336,14 +336,14 @@ func TestBlobFetcherFullDelivery(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 5}, // Force full requests for simplicity &mockRand{value: 5}, // Force full requests for simplicity
) )
}, },
@ -370,7 +370,7 @@ func TestBlobFetcherFullDelivery(t *testing.T) {
}, },
// All alternates should be clean up on delivery // All alternates should be clean up on delivery
doBlobEnqueue{peer: "A", hashes: []common.Hash{testBlobTxHashes[0]}, cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, &halfCustody)}, custody: halfCustody}, doBlobEnqueue{peer: "A", hashes: []common.Hash{testBlobTxHashes[0]}, cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, halfCustody)}, custody: halfCustody},
isBlobScheduled{announces: nil, fetching: nil}, isBlobScheduled{announces: nil, fetching: nil},
isFetching{hashes: nil}, // fetches should be empty after completion isFetching{hashes: nil}, // fetches should be empty after completion
isCompleted{testBlobTxHashes[0]}, isCompleted{testBlobTxHashes[0]},
@ -384,14 +384,14 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 60}, &mockRand{value: 60},
) )
}, },
@ -425,8 +425,8 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"A": {{hash: testBlobTxHashes[0], custody: custody}}, "A": {{hash: testBlobTxHashes[0], custody: custody}},
"B": {{hash: testBlobTxHashes[0], custody: custody}}, "B": {{hash: testBlobTxHashes[0], custody: custody}},
"C": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "C": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
"A": {{hash: testBlobTxHashes[0], custody: custody}}, "A": {{hash: testBlobTxHashes[0], custody: custody}},
@ -438,12 +438,12 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
doDrop("B"), doDrop("B"),
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"C": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "C": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
"C": {{hash: testBlobTxHashes[0], custody: *frontCustody.Intersection(&custody)}}, "C": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
}, },
@ -451,22 +451,22 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
doBlobEnqueue{ doBlobEnqueue{
peer: "C", peer: "C",
hashes: []common.Hash{testBlobTxHashes[0]}, hashes: []common.Hash{testBlobTxHashes[0]},
cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, frontCustody.Intersection(&custody))}, cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, frontCustody.Intersection(custody))},
custody: *frontCustody.Intersection(&custody), custody: frontCustody.Intersection(custody),
}, },
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
}, },
isFetching{ isFetching{
hashes: map[common.Hash]fetchInfo{ hashes: map[common.Hash]fetchInfo{
testBlobTxHashes[0]: { testBlobTxHashes[0]: {
fetching: &custody, fetching: &custody,
fetched: frontCustody.Intersection(&custody).Indices(), fetched: frontCustody.Intersection(custody).Indices(),
}, },
}, },
}, },
@ -475,11 +475,11 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
doBlobNotify{peer: "E", hashes: []common.Hash{testBlobTxHashes[0]}, custody: custody}, doBlobNotify{peer: "E", hashes: []common.Hash{testBlobTxHashes[0]}, custody: custody},
isBlobScheduled{ isBlobScheduled{
announces: map[string][]blobAnnounce{ announces: map[string][]blobAnnounce{
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
"E": {{hash: testBlobTxHashes[0], custody: custody}}, "E": {{hash: testBlobTxHashes[0], custody: custody}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
"D": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "D": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
}, },
@ -490,14 +490,14 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
"E": {{hash: testBlobTxHashes[0], custody: custody}}, "E": {{hash: testBlobTxHashes[0], custody: custody}},
}, },
fetching: map[string][]blobAnnounce{ fetching: map[string][]blobAnnounce{
"E": {{hash: testBlobTxHashes[0], custody: *backCustody.Intersection(&custody)}}, "E": {{hash: testBlobTxHashes[0], custody: backCustody.Intersection(custody)}},
}, },
}, },
isFetching{ isFetching{
hashes: map[common.Hash]fetchInfo{ hashes: map[common.Hash]fetchInfo{
testBlobTxHashes[0]: { testBlobTxHashes[0]: {
fetching: &custody, fetching: &custody,
fetched: frontCustody.Intersection(&custody).Indices(), fetched: frontCustody.Intersection(custody).Indices(),
}, },
}, },
}, },
@ -506,8 +506,8 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
doBlobEnqueue{ doBlobEnqueue{
peer: "E", peer: "E",
hashes: []common.Hash{testBlobTxHashes[0]}, hashes: []common.Hash{testBlobTxHashes[0]},
cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, backCustody.Intersection(&custody))}, cells: [][]kzg4844.Cell{selectCells(testBlobSidecars[0].Cells, backCustody.Intersection(custody))},
custody: *backCustody.Intersection(&custody), custody: backCustody.Intersection(custody),
}, },
isCompleted{testBlobTxHashes[0]}, isCompleted{testBlobTxHashes[0]},
}, },
@ -520,14 +520,14 @@ func TestBlobFetcherAvailabilityTimeout(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 60}, &mockRand{value: 60},
) )
}, },
@ -562,14 +562,14 @@ func TestBlobFetcherPeerDrop(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 5}, &mockRand{value: 5},
) )
}, },
@ -639,14 +639,14 @@ func TestBlobFetcherFetchTimeout(t *testing.T) {
init: func() *BlobFetcher { init: func() *BlobFetcher {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, *types.CustodyBitmap) {}, HasPayload: func(common.Hash) bool { return false }, AddCells: func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 5}, &mockRand{value: 5},
) )
}, },
@ -817,7 +817,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
for _, ann := range announces { for _, ann := range announces {
if cellWithSeq, ok := peerAnnounces[ann.hash]; !ok { if cellWithSeq, ok := peerAnnounces[ann.hash]; !ok {
t.Errorf("step %d, peer %s: hash %x missing from announces", i, peer, ann.hash) t.Errorf("step %d, peer %s: hash %x missing from announces", i, peer, ann.hash)
} else if *cellWithSeq.cells != ann.custody { } else if cellWithSeq.cells != ann.custody {
t.Errorf("step %d, peer %s, hash %x: custody mismatch in announces", i, peer, ann.hash) t.Errorf("step %d, peer %s, hash %x: custody mismatch in announces", i, peer, ann.hash)
} }
} }
@ -854,7 +854,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
found := false found := false
for _, cellReq := range peerRequests { for _, cellReq := range peerRequests {
for _, hash := range cellReq.txs { for _, hash := range cellReq.txs {
if hash == req.hash && *cellReq.cells == req.custody { if hash == req.hash && cellReq.cells == req.custody {
found = true found = true
break break
} }
@ -872,7 +872,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
for _, hash := range cellReq.txs { for _, hash := range cellReq.txs {
found := false found := false
for _, req := range requests { for _, req := range requests {
if req.hash == hash && *cellReq.cells == req.custody { if req.hash == hash && cellReq.cells == req.custody {
found = true found = true
break break
} }
@ -906,7 +906,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
if cellWithSeq := peerAnnounces[hash]; cellWithSeq != nil { if cellWithSeq := peerAnnounces[hash]; cellWithSeq != nil {
if altCustody, ok := alternates[peer]; !ok { if altCustody, ok := alternates[peer]; !ok {
t.Errorf("step %d, hash %x: peer %s missing from alternates", i, hash, peer) t.Errorf("step %d, hash %x: peer %s missing from alternates", i, hash, peer)
} else if *altCustody != *cellWithSeq.cells { } else if altCustody != cellWithSeq.cells {
t.Errorf("step %d, hash %x, peer %s: custody bitmap mismatch in alternates", i, hash, peer) t.Errorf("step %d, hash %x, peer %s: custody bitmap mismatch in alternates", i, hash, peer)
} }
} }
@ -916,7 +916,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
for peer, altCustody := range alternates { for peer, altCustody := range alternates {
if fetcher.announces[peer] == nil || fetcher.announces[peer][hash] == nil { if fetcher.announces[peer] == nil || fetcher.announces[peer][hash] == nil {
t.Errorf("step %d, hash %x: peer %s extra in alternates", i, hash, peer) t.Errorf("step %d, hash %x: peer %s extra in alternates", i, hash, peer)
} else if cellWithSeq := fetcher.announces[peer][hash]; *cellWithSeq.cells != *altCustody { } else if cellWithSeq := fetcher.announces[peer][hash]; cellWithSeq.cells != altCustody {
t.Errorf("step %d, hash %x, peer %s: custody bitmap mismatch between announces and alternates", i, hash, peer) t.Errorf("step %d, hash %x, peer %s: custody bitmap mismatch between announces and alternates", i, hash, peer)
} }
} }
@ -931,9 +931,7 @@ func testBlobFetcher(t *testing.T, tt blobFetcherTest) {
} else { } else {
// Check fetching bitmap // Check fetching bitmap
if expected.fetching != nil { if expected.fetching != nil {
if fetchStatus.fetching == nil { if fetchStatus.fetching != *expected.fetching {
t.Errorf("step %d, hash %x: fetching bitmap is nil", i, hash)
} else if *fetchStatus.fetching != *expected.fetching {
t.Errorf("step %d, hash %x: fetching bitmap mismatch", i, hash) t.Errorf("step %d, hash %x: fetching bitmap mismatch", i, hash)
} }
} }
@ -1027,7 +1025,7 @@ func TestMultiBlobDeliveryVerification(t *testing.T) {
return NewBlobFetcher( return NewBlobFetcher(
BlobFetcherFunctions{ BlobFetcherFunctions{
HasPayload: func(common.Hash) bool { return false }, HasPayload: func(common.Hash) bool { return false },
AddCells: func(h common.Hash, deliveries map[string]*PeerCellDelivery, custody *types.CustodyBitmap) { AddCells: func(h common.Hash, deliveries map[string]*PeerCellDelivery, custody types.CustodyBitmap) {
// Verify each peer's delivered cells pass KZG cell proof verification // Verify each peer's delivered cells pass KZG cell proof verification
for _, d := range deliveries { for _, d := range deliveries {
var cellProofs []kzg4844.Proof var cellProofs []kzg4844.Proof
@ -1039,12 +1037,12 @@ func TestMultiBlobDeliveryVerification(t *testing.T) {
verifyErr = kzg4844.VerifyCells(d.Cells, sidecar.Commitments, cellProofs, d.Indices) verifyErr = kzg4844.VerifyCells(d.Cells, sidecar.Commitments, cellProofs, d.Indices)
} }
}, },
FetchPayloads: func(string, []common.Hash, *types.CustodyBitmap) error { FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error {
return nil return nil
}, },
DropPeer: func(string) {}, DropPeer: func(string) {},
}, },
&custody, custody,
&mockRand{value: 60}, // Force partial requests (60 >= fetchProbability) &mockRand{value: 60}, // Force partial requests (60 >= fetchProbability)
) )
}, },
@ -1065,15 +1063,15 @@ func TestMultiBlobDeliveryVerification(t *testing.T) {
doBlobEnqueue{ doBlobEnqueue{
peer: "D", peer: "D",
hashes: []common.Hash{testBlobTxHashes[0]}, hashes: []common.Hash{testBlobTxHashes[0]},
cells: [][]kzg4844.Cell{selectMultiBlobCells(sidecar, *backCustody.Intersection(&custody))}, cells: [][]kzg4844.Cell{selectMultiBlobCells(sidecar, backCustody.Intersection(custody))},
custody: *backCustody.Intersection(&custody), custody: backCustody.Intersection(custody),
}, },
// Deliver front cells from C // Deliver front cells from C
doBlobEnqueue{ doBlobEnqueue{
peer: "C", peer: "C",
hashes: []common.Hash{testBlobTxHashes[0]}, hashes: []common.Hash{testBlobTxHashes[0]},
cells: [][]kzg4844.Cell{selectMultiBlobCells(sidecar, *frontCustody.Intersection(&custody))}, cells: [][]kzg4844.Cell{selectMultiBlobCells(sidecar, frontCustody.Intersection(custody))},
custody: *frontCustody.Intersection(&custody), custody: frontCustody.Intersection(custody),
}, },
isCompleted{testBlobTxHashes[0]}, isCompleted{testBlobTxHashes[0]},
}, },

6
eth/handler.go
View file

@ -208,7 +208,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
// Construct the blob fetcher for cell-based blob data availability // Construct the blob fetcher for cell-based blob data availability
blobCallbacks := fetcher.BlobFetcherFunctions{ blobCallbacks := fetcher.BlobFetcherFunctions{
FetchPayloads: func(peer string, hashes []common.Hash, cells *types.CustodyBitmap) error { FetchPayloads: func(peer string, hashes []common.Hash, cells types.CustodyBitmap) error {
p := h.peers.peer(peer) p := h.peers.peer(peer)
if p == nil { if p == nil {
return errors.New("unknown peer") return errors.New("unknown peer")
@ -218,7 +218,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
HasPayload: func(hash common.Hash) bool { HasPayload: func(hash common.Hash) bool {
return h.blobpool.Has(hash) || blobBuffer.HasCells(hash) return h.blobpool.Has(hash) || blobBuffer.HasCells(hash)
}, },
AddCells: func(hash common.Hash, deliveries map[string]*fetcher.PeerCellDelivery, custody *types.CustodyBitmap) { AddCells: func(hash common.Hash, deliveries map[string]*fetcher.PeerCellDelivery, custody types.CustodyBitmap) {
converted := make(map[string]*blobpool.PeerDelivery, len(deliveries)) converted := make(map[string]*blobpool.PeerDelivery, len(deliveries))
for peer, d := range deliveries { for peer, d := range deliveries {
converted[peer] = &blobpool.PeerDelivery{Cells: d.Cells, Indices: d.Indices} converted[peer] = &blobpool.PeerDelivery{Cells: d.Cells, Indices: d.Indices}
@ -227,7 +227,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
}, },
DropPeer: h.removePeer, DropPeer: h.removePeer,
} }
h.blobFetcher = fetcher.NewBlobFetcher(blobCallbacks, &config.Custody, nil) h.blobFetcher = fetcher.NewBlobFetcher(blobCallbacks, config.Custody, nil)
return h, nil return h, nil
} }

4
eth/protocols/eth/peer.go
View file

@ -264,7 +264,7 @@ func (p *Peer) ReplyCells(id uint64, hashes []common.Hash, cells [][]kzg4844.Cel
} }
// RequestPayload fetches a batch of cells from a remote node. // RequestPayload fetches a batch of cells from a remote node.
func (p *Peer) RequestPayload(hashes []common.Hash, cell *types.CustodyBitmap) error { func (p *Peer) RequestPayload(hashes []common.Hash, cell types.CustodyBitmap) error {
p.Log().Debug("Fetching batch of cells", "txcount", len(hashes), "cellcount", cell.OneCount()) p.Log().Debug("Fetching batch of cells", "txcount", len(hashes), "cellcount", cell.OneCount())
id := rand.Uint64() id := rand.Uint64()
@ -281,7 +281,7 @@ func (p *Peer) RequestPayload(hashes []common.Hash, cell *types.CustodyBitmap) e
RequestId: id, RequestId: id,
GetCellsRequest: GetCellsRequest{ GetCellsRequest: GetCellsRequest{
Hashes: hashes, Hashes: hashes,
Mask: *cell, Mask: cell,
}, },
}) })
} }