Send raft pending logs events on any subscribed channels. (#1157)

* Send raft pending logs events on any subscribed channels.
Fixes bug introduced with recent upstream geth merges.

* Add tests verifying raft sends events to correct feed and subscribers receive those events

Author: chris-j-h

core/events.go

@@ -24,11 +24,6 @@ import (
 // NewTxsEvent is posted when a batch of transactions enter the transaction pool.
 type NewTxsEvent struct{ Txs []*types.Transaction }
 
-// PendingLogsEvent is posted pre mining and notifies of pending logs.
-type PendingLogsEvent struct {
-	Logs []*types.Log
-}
-
 // PendingStateEvent is posted pre mining and notifies of pending state changes.
 type PendingStateEvent struct{}
 

eth/api_backend.go

@@ -249,7 +249,7 @@ func (b *EthAPIBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEven
 }
 
 func (b *EthAPIBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
-	return b.eth.miner.SubscribePendingLogs(ch)
+	return b.eth.SubscribePendingLogs(ch) // Quorum
 }
 
 func (b *EthAPIBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {

eth/api_backend_test.go

@@ -0,0 +1,63 @@
+package eth
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEthAPIBackend_SubscribePendingLogsEvent(t *testing.T) {
+	conf := &Config{
+		RaftMode: false,
+	}
+	stack, err := node.New(&node.Config{})
+	if err != nil {
+		t.Fatalf("failed to create node, err = %v", err)
+	}
+	eth, err := New(stack, conf)
+	if err != nil {
+		t.Fatalf("failed to create eth service, err = %v", err)
+	}
+
+	b := &EthAPIBackend{
+		eth: eth,
+	}
+
+	ch := make(chan []*types.Log, 1)
+
+	_ = b.SubscribePendingLogsEvent(ch)
+
+	recipientCount := eth.ConsensusServicePendingLogsFeed().Send([]*types.Log{})
+
+	require.Zero(t, recipientCount, "not using consensus service so its event feed should not have subscribers")
+	require.Zero(t, len(ch), "not using consensus service so subscribed channel should not have received event")
+}
+
+func TestEthAPIBackend_SubscribePendingLogsEvent_SubscribesToConsensusServiceFeed(t *testing.T) {
+	conf := &Config{
+		RaftMode: true,
+	}
+	stack, err := node.New(&node.Config{})
+	if err != nil {
+		t.Fatalf("failed to create node, err = %v", err)
+	}
+	eth, err := New(stack, conf)
+	if err != nil {
+		t.Fatalf("failed to create eth service, err = %v ", err)
+	}
+
+	b := &EthAPIBackend{
+		eth: eth,
+	}
+
+	ch := make(chan []*types.Log, 1)
+
+	_ = b.SubscribePendingLogsEvent(ch)
+
+	recipientCount := eth.ConsensusServicePendingLogsFeed().Send([]*types.Log{})
+
+	require.NotZero(t, recipientCount, "consensus service in use so its event feed should have subscribers")
+	require.Equal(t, 1, len(ch), "consensus service in use so subscribed channel should have received event")
+}

eth/backend.go

@@ -94,6 +94,9 @@ type Ethereum struct {
 	// Quorum - Multitenancy
 	// contractAuthzProvider is set after node starts instead in New()
 	contractAuthzProvider multitenancy.ContractAuthorizationProvider
+
+	// Quorum - consensus as eth-service (e.g. raft)
+	consensusServicePendingLogsFeed *event.Feed
 }
 
 // Quorum
@@ -153,18 +156,19 @@ func New(stack *node.Node, config *Config) (*Ethereum, error) {
 	}
 
 	eth := &Ethereum{
-		config:            config,
-		chainDb:           chainDb,
-		eventMux:          stack.EventMux(),
-		accountManager:    stack.AccountManager(),
-		engine:            CreateConsensusEngine(stack, chainConfig, config, config.Miner.Notify, config.Miner.Noverify, chainDb),
-		closeBloomHandler: make(chan struct{}),
-		networkID:         config.NetworkId,
-		gasPrice:          config.Miner.GasPrice,
-		etherbase:         config.Miner.Etherbase,
-		bloomRequests:     make(chan chan *bloombits.Retrieval),
-		bloomIndexer:      NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms),
-		p2pServer:         stack.Server(),
+		config:                          config,
+		chainDb:                         chainDb,
+		eventMux:                        stack.EventMux(),
+		accountManager:                  stack.AccountManager(),
+		engine:                          CreateConsensusEngine(stack, chainConfig, config, config.Miner.Notify, config.Miner.Noverify, chainDb),
+		closeBloomHandler:               make(chan struct{}),
+		networkID:                       config.NetworkId,
+		gasPrice:                        config.Miner.GasPrice,
+		etherbase:                       config.Miner.Etherbase,
+		bloomRequests:                   make(chan chan *bloombits.Retrieval),
+		bloomIndexer:                    NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms),
+		p2pServer:                       stack.Server(),
+		consensusServicePendingLogsFeed: new(event.Feed),
 	}
 
 	// Quorum: Set protocol Name/Version
@@ -637,3 +641,18 @@ func (s *Ethereum) Stop() error {
 func (s *Ethereum) CalcGasLimit(block *types.Block) uint64 {
 	return core.CalcGasLimit(block, s.config.Miner.GasFloor, s.config.Miner.GasCeil)
 }
+
+// (Quorum)
+// ConsensusServicePendingLogsFeed returns an event.Feed.  When the consensus protocol does not use eth.worker (e.g. raft), the event.Feed should be used to send logs from transactions included in the pending block
+func (s *Ethereum) ConsensusServicePendingLogsFeed() *event.Feed {
+	return s.consensusServicePendingLogsFeed
+}
+
+// (Quorum)
+// SubscribePendingLogs starts delivering logs from transactions included in the consensus engine's pending block to the given channel.
+func (s *Ethereum) SubscribePendingLogs(ch chan<- []*types.Log) event.Subscription {
+	if s.config.RaftMode {
+		return s.consensusServicePendingLogsFeed.Subscribe(ch)
+	}
+	return s.miner.SubscribePendingLogs(ch)
+}

raft/backend.go

@@ -35,6 +35,8 @@ type RaftService struct {
 	minter           *minter
 	nodeKey          *ecdsa.PrivateKey
 	calcGasLimitFunc func(block *types.Block) uint64
+
+	pendingLogsFeed *event.Feed
 }
 
 func New(stack *node.Node, chainConfig *params.ChainConfig, raftId, raftPort uint16, joinExisting bool, blockTime time.Duration, e *eth.Ethereum, startPeers []*enode.Node, datadir string, useDns bool) (*RaftService, error) {
@@ -48,6 +50,7 @@ func New(stack *node.Node, chainConfig *params.ChainConfig, raftId, raftPort uin
 		startPeers:       startPeers,
 		nodeKey:          stack.GetNodeKey(),
 		calcGasLimitFunc: e.CalcGasLimit,
+		pendingLogsFeed:  e.ConsensusServicePendingLogsFeed(),
 	}
 
 	service.minter = newMinter(chainConfig, service, blockTime)

raft/backend_test.go

@@ -0,0 +1,42 @@
+package raft
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/eth"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_New_RegistersEthServicePendingLogsFeed(t *testing.T) {
+	conf := &eth.Config{
+		RaftMode: true,
+	}
+	stack, err := node.New(&node.Config{})
+	if err != nil {
+		t.Fatalf("failed to create node, err = %v", err)
+	}
+	ethService, err := eth.New(stack, conf)
+	if err != nil {
+		t.Fatalf("failed to create eth service, err = %v", err)
+	}
+
+	tmpWorkingDir, err := ioutil.TempDir("", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		_ = os.RemoveAll(tmpWorkingDir)
+	}()
+
+	raftService, err := New(stack, &params.ChainConfig{}, 0, 0, false, time.Second, ethService, nil, tmpWorkingDir, false)
+	if err != nil {
+		t.Fatalf("failed to create raft service, err = %v", err)
+	}
+
+	require.Equal(t, ethService.ConsensusServicePendingLogsFeed(), raftService.pendingLogsFeed, "raft service has not been set up with Ethereum service's consensusServicePendingLogsFeed")
+}

raft/minter.go

@@ -301,7 +301,7 @@ func (minter *minter) firePendingBlockEvents(logs []*types.Log) {
 	}
 
 	go func() {
-		minter.mux.Post(core.PendingLogsEvent{Logs: copiedLogs})
+		minter.eth.pendingLogsFeed.Send(copiedLogs)
 		minter.mux.Post(core.PendingStateEvent{})
 	}()
 }