Fix for RPC filters, remove mined transactions (#3142)

Co-authored-by: Alexey Sharp <alexeysharp@Alexeys-iMac.local>

Author: AlexeyAkhunov

cmd/rpcdaemon/commands/eth_filters.go

@@ -95,9 +95,11 @@ func (api *APIImpl) NewPendingTransactions(ctx context.Context) (*rpc.Subscripti
 			select {
 			case txs := <-txsCh:
 				for _, t := range txs {
-					err := notifier.Notify(rpcSub.ID, t.Hash())
-					if err != nil {
-						log.Warn("error while notifying subscription", "err", err)
+					if t != nil {
+						err := notifier.Notify(rpcSub.ID, t.Hash())
+						if err != nil {
+							log.Warn("error while notifying subscription", "err", err)
+						}
 					}
 				}
 			case <-rpcSub.Err():

cmd/rpcdaemon/filters/filters.go

@@ -222,9 +222,12 @@ func (ff *Filters) subscribeToPendingBlocks(ctx context.Context, mining txpool.M
 }
 
 func (ff *Filters) HandlePendingBlock(reply *txpool.OnPendingBlockReply) {
+	if len(reply.RplBlock) == 0 {
+		return
+	}
 	b := &types.Block{}
 	if err := rlp.Decode(bytes.NewReader(reply.RplBlock), b); err != nil {
-		log.Warn("OnNewTx rpc filters, unprocessable payload", "err", err)
+		log.Warn("HandlePendingBlock rpc filters, unprocessable payload", "err", err)
 	}
 
 	ff.mu.Lock()
@@ -263,9 +266,12 @@ func (ff *Filters) subscribeToPendingLogs(ctx context.Context, mining txpool.Min
 }
 
 func (ff *Filters) HandlePendingLogs(reply *txpool.OnPendingLogsReply) {
+	if len(reply.RplLogs) == 0 {
+		return
+	}
 	l := []*types.Log{}
 	if err := rlp.Decode(bytes.NewReader(reply.RplLogs), &l); err != nil {
-		log.Warn("OnNewTx rpc filters, unprocessable payload", "err", err)
+		log.Warn("HandlePendingLogs rpc filters, unprocessable payload", "err", err)
 	}
 
 	ff.mu.RLock()
@@ -338,6 +344,9 @@ func (ff *Filters) OnNewEvent(event *remote.SubscribeReply) {
 	switch event.Type {
 	case remote.Event_HEADER:
 		payload := event.Data
+		if len(payload) == 0 {
+			return
+		}
 		var header types.Header
 
 		err := rlp.Decode(bytes.NewReader(payload), &header)
@@ -385,6 +394,9 @@ func (ff *Filters) OnNewTx(reply *txpool.OnAddReply) {
 
 	txs := make([]types.Transaction, len(reply.RplTxs))
 	for i, rlpTx := range reply.RplTxs {
+		if len(rlpTx) == 0 {
+			continue
+		}
 		var decodeErr error
 		s := rlp.NewStream(bytes.NewReader(rlpTx), uint64(len(rlpTx)))
 		txs[i], decodeErr = types.DecodeTransaction(s)

go.mod

@@ -36,7 +36,7 @@ require (
 	github.com/json-iterator/go v1.1.12
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/kevinburke/go-bindata v3.21.0+incompatible
-	github.com/ledgerwatch/erigon-lib v0.0.0-20211216220419-44c18ee5f94a
+	github.com/ledgerwatch/erigon-lib v0.0.0-20211217093552-7d82c6ac76e6
 	github.com/ledgerwatch/log/v3 v3.4.0
 	github.com/ledgerwatch/secp256k1 v1.0.0
 	github.com/logrusorgru/aurora/v3 v3.0.0

go.sum

@@ -500,8 +500,8 @@ github.com/kylelemons/godebug v0.0.0-20170224010052-a616ab194758 h1:0D5M2HQSGD3P
 github.com/kylelemons/godebug v0.0.0-20170224010052-a616ab194758/go.mod h1:B69LEHPfb2qLo0BaaOLcbitczOKLWTsrBG9LczfCD4k=
 github.com/leanovate/gopter v0.2.9 h1:fQjYxZaynp97ozCzfOyOuAGOU4aU/z37zf/tOujFk7c=
 github.com/leanovate/gopter v0.2.9/go.mod h1:U2L/78B+KVFIx2VmW6onHJQzXtFb+p5y3y2Sh+Jxxv8=
-github.com/ledgerwatch/erigon-lib v0.0.0-20211216220419-44c18ee5f94a h1:saAP2TrAu8ifG6CJ30GAs5ydlaUtuVSmp0xe7clejIE=
-github.com/ledgerwatch/erigon-lib v0.0.0-20211216220419-44c18ee5f94a/go.mod h1:lyGP3i0x4CeabdKZ4beycD5xZfHWZwJsAX+70OfGj4Y=
+github.com/ledgerwatch/erigon-lib v0.0.0-20211217093552-7d82c6ac76e6 h1:xXbeZ6gQLwPSNE8o6W99hZqjDGwN95fF7Q2p3zmRfAE=
+github.com/ledgerwatch/erigon-lib v0.0.0-20211217093552-7d82c6ac76e6/go.mod h1:lyGP3i0x4CeabdKZ4beycD5xZfHWZwJsAX+70OfGj4Y=
 github.com/ledgerwatch/log/v3 v3.4.0 h1:SEIOcv5a2zkG3PmoT5jeTU9m/0nEUv0BJS5bzsjwKCI=
 github.com/ledgerwatch/log/v3 v3.4.0/go.mod h1:VXcz6Ssn6XEeU92dCMc39/g1F0OYAjw1Mt+dGP5DjXY=
 github.com/ledgerwatch/secp256k1 v1.0.0 h1:Usvz87YoTG0uePIV8woOof5cQnLXGYa162rFf3YnwaQ=