Implement Horizontally Sharded Blob Mempool

eth/protocols/eth/handler_test.go

@@ -688,14 +688,26 @@ func testGetPooledTransaction(t *testing.T, blobTx bool) {
 		}
 	}
 
-	// Send the hash request and verify the response
+	hashes := []common.Hash{tx.Hash()}
+
 	p2p.Send(peer.app, GetPooledTransactionsMsg, GetPooledTransactionsPacket{
 		RequestId:                    123,
-		GetPooledTransactionsRequest: []common.Hash{tx.Hash()},
+		GetPooledTransactionsRequest: hashes,
 	})
+
+	var expectedTxs []*types.Transaction
+	peerIDLast4 := peer.ID()[len(peer.ID())-1] & 0x0F
+
+	for _, tx := range []*types.Transaction{tx} {
+		txHashLast4 := tx.Hash().Bytes()[len(tx.Hash().Bytes())-1] & 0x0F
+		if peerIDLast4 != txHashLast4 {
+			expectedTxs = append(expectedTxs, tx)
+		}
+	}
+
 	if err := p2p.ExpectMsg(peer.app, PooledTransactionsMsg, PooledTransactionsPacket{
 		RequestId:                  123,
-		PooledTransactionsResponse: []*types.Transaction{tx},
+		PooledTransactionsResponse: expectedTxs,
 	}); err != nil {
 		t.Errorf("pooled transaction mismatch: %v", err)
 	}

eth/protocols/eth/handlers.go

@@ -514,16 +514,36 @@ func handlePooledTransactions(backend Backend, msg Decoder, peer *Peer) error {
 	if err := msg.Decode(&txs); err != nil {
 		return err
 	}
+	peerID := peer.ID()
+	peerIDLast4 := peerID[len(peerID)-1] & 0x0F
+
+	var filteredTxs []*types.Transaction
+
 	for i, tx := range txs.PooledTransactionsResponse {
 		// Validate and mark the remote transaction
 		if tx == nil {
 			return fmt.Errorf("PooledTransactions: transaction %d is nil", i)
 		}
+		txHash := tx.Hash()
+		txHashLast4 := txHash[len(txHash)-1] & 0x0F
+
+		if txHashLast4 != peerIDLast4 && tx.Type() == types.BlobTxType {
+			continue
+		}
+
 		peer.markTransaction(tx.Hash())
+		filteredTxs = append(filteredTxs, tx)
 	}
 	requestTracker.Fulfil(peer.id, peer.version, PooledTransactionsMsg, txs.RequestId)
 
-	return backend.Handle(peer, &txs.PooledTransactionsResponse)
+	if len(filteredTxs) == 0 {
+		return nil
+	}
+
+	return backend.Handle(peer, &PooledTransactionsPacket{
+		PooledTransactionsResponse: filteredTxs,
+		RequestId:                  txs.RequestId,
+	})
 }
 
 func handleBlockRangeUpdate(backend Backend, msg Decoder, peer *Peer) error {