From 1d59ebae2230c74e957e84461096a26af22d566f Mon Sep 17 00:00:00 2001
From: Arunima Chaudhuri <tinniarunima66@gmail.com>
Date: Tue, 10 Jun 2025 02:31:45 +0530
Subject: [PATCH] Implement Horizontally Sharded Blob Mempool

---
 eth/catalyst/api.go           | 17 +++++++++++++++++
 eth/protocols/eth/handlers.go | 24 +++++++++++++++++++++++-
 2 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go
index 40dae3f6f2..fb2cb2f77f 100644
--- a/eth/catalyst/api.go
+++ b/eth/catalyst/api.go
@@ -150,12 +150,15 @@ type ConsensusAPI struct {
 
 	forkchoiceLock sync.Mutex // Lock for the forkChoiceUpdated method
 	newPayloadLock sync.Mutex // Lock for the NewPayload method
+
+	isProposer atomic.Bool // Tracks if the node is currently identified as a proposer
 }
 
 // NewConsensusAPI creates a new consensus api for the given backend.
 // The underlying blockchain needs to have a valid terminal total difficulty set.
 func NewConsensusAPI(eth *eth.Ethereum) *ConsensusAPI {
 	api := newConsensusAPIWithoutHeartbeat(eth)
+	api.isProposer.Store(false)
 	go api.heartbeat()
 	return api
 }
@@ -172,6 +175,7 @@ func newConsensusAPIWithoutHeartbeat(eth *eth.Ethereum) *ConsensusAPI {
 		invalidBlocksHits: make(map[common.Hash]int),
 		invalidTipsets:    make(map[common.Hash]*types.Header),
 	}
+	api.isProposer.Store(false)
 	eth.Downloader().SetBadBlockCallback(api.setInvalidAncestor)
 	return api
 }
@@ -198,6 +202,9 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV1(update engine.ForkchoiceStateV1, pa
 		case !api.checkFork(payloadAttributes.Timestamp, forks.Paris, forks.Shanghai):
 			return engine.STATUS_INVALID, paramsErr("fcuV1 called post-shanghai")
 		}
+		api.isProposer.Store(true)
+	} else {
+		api.isProposer.Store(false)
 	}
 	return api.forkchoiceUpdated(update, payloadAttributes, engine.PayloadV1, false)
 }
@@ -216,6 +223,9 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV2(update engine.ForkchoiceStateV1, pa
 		case !api.checkFork(params.Timestamp, forks.Paris, forks.Shanghai):
 			return engine.STATUS_INVALID, unsupportedForkErr("fcuV2 must only be called with paris or shanghai payloads")
 		}
+		api.isProposer.Store(true)
+	} else {
+		api.isProposer.Store(false)
 	}
 	return api.forkchoiceUpdated(update, params, engine.PayloadV2, false)
 }
@@ -232,6 +242,9 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV3(update engine.ForkchoiceStateV1, pa
 		case !api.checkFork(params.Timestamp, forks.Cancun, forks.Prague):
 			return engine.STATUS_INVALID, unsupportedForkErr("fcuV3 must only be called for cancun or prague payloads")
 		}
+		api.isProposer.Store(true)
+	} else {
+		api.isProposer.Store(false)
 	}
 	// TODO(matt): the spec requires that fcu is applied when called on a valid
 	// hash, even if params are wrong. To do this we need to split up
@@ -240,6 +253,10 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV3(update engine.ForkchoiceStateV1, pa
 	return api.forkchoiceUpdated(update, params, engine.PayloadV3, false)
 }
 
+func (api *ConsensusAPI) IsProposer() bool {
+	return api.isProposer.Load()
+}
+
 func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payloadAttributes *engine.PayloadAttributes, payloadVersion engine.PayloadVersion, payloadWitness bool) (engine.ForkChoiceResponse, error) {
 	api.forkchoiceLock.Lock()
 	defer api.forkchoiceLock.Unlock()
diff --git a/eth/protocols/eth/handlers.go b/eth/protocols/eth/handlers.go
index 15ad048bcf..740daf6fb0 100644
--- a/eth/protocols/eth/handlers.go
+++ b/eth/protocols/eth/handlers.go
@@ -514,16 +514,38 @@ 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
+
+	
+	isProposer := false
+	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 && !isProposer && 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 {