Txpool upgrades for EIP-4844 Blob Transactions (#1075)

EIP-4844 proposed the new (type-3) Blob Transactions which are expensive
(because blobs are big etc.). The tx-pool should be tightened to
alleviate spam and dos attacks by way of these new transactions.

### Changes
**Keep blob txs cached**
During new block, don't delete mined blobTxs, instead cache it till they
are finalized. This is to avoid fetching them again for unwind. During
unwind, if it's a previously seen blobTxn, try to fetch it from memory.

**Pool config**
BlobSlots: Similar to AccountSlots - limits how many blobs a single
account is allowed to have until it's marked as spamming.
Blob txn priceBump: It's an added field in pool config (with a flag).
Defaults to 100%, since blob txs of equal fee are costlier to
process/execute

---------

Co-authored-by: alex.sharov <AskAlexSharov@gmail.com>

go.work.sum

@@ -0,0 +1,2 @@
+github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
+google.golang.org/genproto v0.0.0-20230526161137-0005af68ea54 h1:9NWlQfY2ePejTmfwUH1OWwmznFa+0kKcHGPDvcPza9M=

txpool/fetch.go

@@ -266,6 +266,7 @@ func (f *Fetch) handleInboundMessage(ctx context.Context, req *sentry.InboundMes
 				unknownHashes = append(unknownHashes, hashes[i:i+32]...)
 			}
 		}
+
 		if len(unknownHashes) > 0 {
 			var encodedRequest []byte
 			var messageID sentry.MessageId
@@ -284,30 +285,24 @@ func (f *Fetch) handleInboundMessage(ctx context.Context, req *sentry.InboundMes
 		//TODO: handleInboundMessage is single-threaded - means it can accept as argument couple buffers (or analog of txParseContext). Protobuf encoding will copy data anyway, but DirectClient doesn't
 		var encodedRequest []byte
 		var messageID sentry.MessageId
-		switch req.Id {
-		case sentry.MessageId_GET_POOLED_TRANSACTIONS_66:
-			messageID = sentry.MessageId_POOLED_TRANSACTIONS_66
-			requestID, hashes, _, err := types2.ParseGetPooledTransactions66(req.Data, 0, nil)
+		messageID = sentry.MessageId_POOLED_TRANSACTIONS_66
+		requestID, hashes, _, err := types2.ParseGetPooledTransactions66(req.Data, 0, nil)
+		if err != nil {
+			return err
+		}
+		_ = requestID
+		var txs [][]byte
+		for i := 0; i < len(hashes); i += 32 {
+			txn, err := f.pool.GetRlp(tx, hashes[i:i+32])
 			if err != nil {
 				return err
 			}
-			_ = requestID
-			var txs [][]byte
-			for i := 0; i < len(hashes); i += 32 {
-				txn, err := f.pool.GetRlp(tx, hashes[i:i+32])
-				if err != nil {
-					return err
-				}
-				if txn == nil {
-					continue
-				}
-				txs = append(txs, txn)
+			if txn == nil {
+				continue
 			}
-
-			encodedRequest = types2.EncodePooledTransactions66(txs, requestID, nil)
-		default:
-			return fmt.Errorf("unexpected message: %s", req.Id.String())
+			txs = append(txs, txn)
 		}
+		encodedRequest = types2.EncodePooledTransactions66(txs, requestID, nil)
 
 		if _, err := sentryClient.SendMessageById(f.ctx, &sentry.SendMessageByIdRequest{
 			Data:   &sentry.OutboundMessageData{Id: messageID, Data: encodedRequest},
@@ -452,6 +447,11 @@ func (f *Fetch) handleStateChanges(ctx context.Context, client StateChangesClien
 	if err != nil {
 		return err
 	}
+	tx, err := f.db.BeginRo(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
 	for req, err := stream.Recv(); ; req, err = stream.Recv() {
 		if err != nil {
 			return err
@@ -481,6 +481,12 @@ func (f *Fetch) handleStateChanges(ctx context.Context, client StateChangesClien
 					unwindTxs.Txs[i] = &types2.TxSlot{}
 					if err = f.threadSafeParseStateChangeTxn(func(parseContext *types2.TxParseContext) error {
 						_, err = parseContext.ParseTransaction(change.Txs[i], 0, unwindTxs.Txs[i], unwindTxs.Senders.At(i), false /* hasEnvelope */, false /* wrappedWithBlobs */, nil)
+						if unwindTxs.Txs[i].Type == types2.BlobTxType {
+							knownBlobTxn := f.pool.GetKnownBlobTxn(tx, unwindTxs.Txs[i].IDHash[:])
+							if knownBlobTxn != nil {
+								unwindTxs.Txs[i] = knownBlobTxn.Tx
+							}
+						}
 						return err
 					}); err != nil && !errors.Is(err, context.Canceled) {
 						f.logger.Warn("stream.Recv", "err", err)
@@ -501,3 +507,21 @@ func (f *Fetch) handleStateChanges(ctx context.Context, client StateChangesClien
 		}
 	}
 }
+
+// func (f *Fetch) requestUnknownTxs(unknownHashes types2.Hashes, sentryClient sentry.SentryClient, PeerId *types.H512) error{
+// 	if len(unknownHashes) > 0 {
+// 		var encodedRequest []byte
+// 		var err error
+// 		var messageID sentry.MessageId
+// 		if encodedRequest, err = types2.EncodeGetPooledTransactions66(unknownHashes, uint64(1), nil); err != nil {
+// 			return err
+// 		}
+// 		messageID = sentry.MessageId_GET_POOLED_TRANSACTIONS_66
+// 		if _, err := sentryClient.SendMessageById(f.ctx, &sentry.SendMessageByIdRequest{
+// 			Data:   &sentry.OutboundMessageData{Id: messageID, Data: encodedRequest},
+// 			PeerId: PeerId,
+// 		}, &grpc.EmptyCallOption{}); err != nil {
+// 			return err
+// 		}
+// 	}
+// }