From 6ae8387e1a45a407fdb12a2c30dc368fa001b20e Mon Sep 17 00:00:00 2001 From: meows Date: Mon, 11 Jan 2021 18:13:48 -0600 Subject: [PATCH 1/4] eth/filters,ethclient,node: install newSideHeads subscription The newSideHeads subscription work very similarly to the newHeads subscription; instead, non-canonical blocks are channeled. --- accounts/abi/bind/backends/simulated.go | 31 +++++ core/blockchain.go | 2 +- eth/filters/api.go | 68 ++++++++++- eth/filters/filter.go | 1 + eth/filters/filter_system.go | 52 ++++++++- eth/filters/filter_system_test.go | 61 ++++++++++ ethclient/ethclient.go | 6 + ethclient/ethclient_test.go | 144 ++++++++++++++++++++++++ node/openrpc.go | 1 + 9 files changed, 362 insertions(+), 4 deletions(-) diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 50504196a2..0174f8c77c 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -677,6 +677,33 @@ func (b *SimulatedBackend) SubscribeNewHead(ctx context.Context, ch chan<- *type }), nil } +// SubscribeNewHead returns an event subscription for a new header imported as non-canonical (side status). +func (b *SimulatedBackend) SubscribeNewSideHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { + // subscribe to a new head + sink := make(chan *types.Header) + sub := b.events.SubscribeNewSideHeads(sink) + + return event.NewSubscription(func(quit <-chan struct{}) error { + defer sub.Unsubscribe() + for { + select { + case head := <-sink: + select { + case ch <- head: + case err := <-sub.Err(): + return err + case <-quit: + return nil + } + case err := <-sub.Err(): + return err + case <-quit: + return nil + } + } + }), nil +} + // AdjustTime adds a time shift to the simulated clock. // It can only be called on empty blocks. func (b *SimulatedBackend) AdjustTime(adjustment time.Duration) error { @@ -770,6 +797,10 @@ func (fb *filterBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Su return fb.bc.SubscribeChainEvent(ch) } +func (fb *filterBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription { + return fb.bc.SubscribeChainSideEvent(ch) +} + func (fb *filterBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription { return fb.bc.SubscribeRemovedLogsEvent(ch) } diff --git a/core/blockchain.go b/core/blockchain.go index fec34b226d..4f2e1dae76 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1694,7 +1694,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. } // In theory we should fire a ChainHeadEvent when we inject // a canonical block, but sometimes we can insert a batch of - // canonicial blocks. Avoid firing too much ChainHeadEvents, + // canonical blocks. Avoid firing too much ChainHeadEvents, // we will fire an accumulated ChainHeadEvent and disable fire // event here. if emitHeadEvent { diff --git a/eth/filters/api.go b/eth/filters/api.go index 30d7b71c31..0221bc962c 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -203,6 +203,40 @@ func (api *PublicFilterAPI) NewBlockFilter() rpc.ID { return headerSub.ID } +// NewSideBlockFilter creates a filter that fetches blocks that are imported into the chain with a non-canonical status. +// It is part of the filter package since polling goes with eth_getFilterChanges. +func (api *PublicFilterAPI) NewSideBlockFilter() rpc.ID { + var ( + headers = make(chan *types.Header) + headerSub = api.events.SubscribeNewSideHeads(headers) + ) + + api.filtersMu.Lock() + api.filters[headerSub.ID] = &filter{typ: SideBlocksSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: headerSub} + api.filtersMu.Unlock() + + go func() { + for { + select { + case h := <-headers: + api.filtersMu.Lock() + if f, found := api.filters[headerSub.ID]; found { + f.hashes = append(f.hashes, h.Hash()) + } + api.filtersMu.Unlock() + case <-headerSub.Err(): + api.filtersMu.Lock() + delete(api.filters, headerSub.ID) + api.filtersMu.Unlock() + return + } + } + }() + + return headerSub.ID +} + + // NewHeads send a notification each time a new (header) block is appended to the chain. func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) @@ -233,7 +267,37 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er return rpcSub, nil } -// Logs creates a subscription that fires for all new log that match the given filter criteria. +// NewHeads send a notification each time a new (header) block is appended to the chain. +func (api *PublicFilterAPI) NewSideHeads(ctx context.Context) (*rpc.Subscription, error) { + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported + } + + rpcSub := notifier.CreateSubscription() + + go func() { + headers := make(chan *types.Header) + headersSub := api.events.SubscribeNewSideHeads(headers) + + for { + select { + case h := <-headers: + notifier.Notify(rpcSub.ID, h) + case <-rpcSub.Err(): + headersSub.Unsubscribe() + return + case <-notifier.Closed(): + headersSub.Unsubscribe() + return + } + } + }() + + return rpcSub, nil +} + +// Logs creates a subscription that fires for all new logs that match the given filter criteria. func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { @@ -424,7 +488,7 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) { f.deadline.Reset(deadline) switch f.typ { - case PendingTransactionsSubscription, BlocksSubscription: + case PendingTransactionsSubscription, BlocksSubscription, SideBlocksSubscription: hashes := f.hashes f.hashes = nil return returnHashes(hashes), nil diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 17635837af..33499e47af 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -39,6 +39,7 @@ type Backend interface { SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription + SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index a105ec51c3..bc29323f07 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -52,6 +52,8 @@ const ( PendingTransactionsSubscription // BlocksSubscription queries hashes for blocks that are imported BlocksSubscription + // SideBlocksSubscription queries blocks that are imported non-canonically + SideBlocksSubscription // LastSubscription keeps track of the last index LastIndexSubscription ) @@ -93,6 +95,7 @@ type EventSystem struct { rmLogsSub event.Subscription // Subscription for removed log event pendingLogsSub event.Subscription // Subscription for pending log event chainSub event.Subscription // Subscription for new chain event + chainSideSub event.Subscription // Subscription for new side chain event // Channels install chan *subscription // install filter for event notification @@ -102,6 +105,7 @@ type EventSystem struct { pendingLogsCh chan []*types.Log // Channel to receive new log event rmLogsCh chan core.RemovedLogsEvent // Channel to receive removed log event chainCh chan core.ChainEvent // Channel to receive new chain event + chainSideCh chan core.ChainSideEvent // Channel to receive new side chain event } // NewEventSystem creates a new manager that listens for event on the given mux, @@ -121,6 +125,7 @@ func NewEventSystem(backend Backend, lightMode bool) *EventSystem { rmLogsCh: make(chan core.RemovedLogsEvent, rmLogsChanSize), pendingLogsCh: make(chan []*types.Log, logsChanSize), chainCh: make(chan core.ChainEvent, chainEvChanSize), + chainSideCh: make(chan core.ChainSideEvent, chainEvChanSize), } // Subscribe events @@ -128,10 +133,11 @@ func NewEventSystem(backend Backend, lightMode bool) *EventSystem { m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh) m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh) m.chainSub = m.backend.SubscribeChainEvent(m.chainCh) + m.chainSideSub = m.backend.SubscribeChainSideEvent(m.chainSideCh) m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh) // Make sure none of the subscriptions are empty - if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.pendingLogsSub == nil { + if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.chainSideSub == nil || m.pendingLogsSub == nil { log.Crit("Subscribe for event system failed") } @@ -290,6 +296,22 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti return es.subscribe(sub) } +// SubscribeNewSideHeads creates a subscription that writes the header of a block that is +// imported as a side chain. +func (es *EventSystem) SubscribeNewSideHeads(headers chan *types.Header) *Subscription { + sub := &subscription{ + id: rpc.NewID(), + typ: SideBlocksSubscription, + created: time.Now(), + logs: make(chan []*types.Log), + hashes: make(chan []common.Hash), + headers: headers, + installed: make(chan struct{}), + err: make(chan error), + } + return es.subscribe(sub) +} + // SubscribePendingTxs creates a subscription that writes transaction hashes for // transactions that enter the transaction pool. func (es *EventSystem) SubscribePendingTxs(hashes chan []common.Hash) *Subscription { @@ -366,6 +388,25 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) } } +func (es *EventSystem) handleChainSideEvent(filters filterIndex, ev core.ChainSideEvent) { + for _, f := range filters[SideBlocksSubscription] { + f.headers <- ev.Block.Header() + } + // Handle filtered log eventing similarly to the newHead event, except that 'remove' will always be set to true + // (indicating the logs come from a non-canonical block). + // When newHeads and newSideHeads are subscribed to at the same time, this can result in certain logs being broadcast + // repetitiously. + if es.lightMode && len(filters[LogsSubscription]) > 0 { + es.lightFilterNewSideHead(ev.Block.Header(), func(header *types.Header, remove bool) { + for _, f := range filters[LogsSubscription] { + if matchedLogs := es.lightFilterLogs(header, f.logsCrit.Addresses, f.logsCrit.Topics, remove); len(matchedLogs) > 0 { + f.logs <- matchedLogs + } + } + }) + } +} + func (es *EventSystem) lightFilterNewHead(newHeader *types.Header, callBack func(*types.Header, bool)) { oldh := es.lastHead es.lastHead = newHeader @@ -399,6 +440,10 @@ func (es *EventSystem) lightFilterNewHead(newHeader *types.Header, callBack func } } +func (es *EventSystem) lightFilterNewSideHead(header *types.Header, callBack func(*types.Header, bool)) { + callBack(header, true) +} + // filter logs of a single header in light client mode func (es *EventSystem) lightFilterLogs(header *types.Header, addresses []common.Address, topics [][]common.Hash, remove bool) []*types.Log { if bloomFilter(header.Bloom, addresses, topics) { @@ -448,6 +493,7 @@ func (es *EventSystem) eventLoop() { es.rmLogsSub.Unsubscribe() es.pendingLogsSub.Unsubscribe() es.chainSub.Unsubscribe() + es.chainSideSub.Unsubscribe() }() index := make(filterIndex) @@ -467,6 +513,8 @@ func (es *EventSystem) eventLoop() { es.handlePendingLogs(index, ev) case ev := <-es.chainCh: es.handleChainEvent(index, ev) + case ev := <-es.chainSideCh: + es.handleChainSideEvent(index, ev) case f := <-es.install: if f.typ == MinedAndPendingLogsSubscription { @@ -497,6 +545,8 @@ func (es *EventSystem) eventLoop() { return case <-es.chainSub.Err(): return + case <-es.chainSideSub.Err(): + return } } } diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index cc37b2e17d..732744cb81 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -49,6 +49,7 @@ type testBackend struct { rmLogsFeed event.Feed pendingLogsFeed event.Feed chainFeed event.Feed + chainSideFeed event.Feed } func (b *testBackend) ChainDb() ethdb.Database { @@ -123,6 +124,10 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc return b.chainFeed.Subscribe(ch) } +func (b *testBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription { + return b.chainSideFeed.Subscribe(ch) +} + func (b *testBackend) BloomStatus() (uint64, uint64) { return vars.BloomBitsBlocks, b.sections } @@ -210,6 +215,62 @@ func TestBlockSubscription(t *testing.T) { <-sub1.Err() } +// TestSideBlockSubscription tests if a block subscription returns block hashes for posted chain events. +// It creates multiple subscriptions: +// - one at the start and should receive all posted chain events and a second (blockHashes) +// - one that is created after a cutoff moment and uninstalled after a second cutoff moment (blockHashes[cutoff1:cutoff2]) +// - one that is created after the second cutoff moment (blockHashes[cutoff2:]) +func TestSideBlockSubscription(t *testing.T) { + t.Parallel() + + var ( + db = rawdb.NewMemoryDatabase() + backend = &testBackend{db: db} + api = NewPublicFilterAPI(backend, false) + genesis = core.MustCommitGenesis(db, new(genesisT.Genesis)) + chain, _ = core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), db, 10, func(i int, gen *core.BlockGen) {}) + chainSideEvents = []core.ChainSideEvent{} + ) + + for _, blk := range chain { + chainSideEvents = append(chainSideEvents, core.ChainSideEvent{Block: blk}) + } + + chan0 := make(chan *types.Header) + sub0 := api.events.SubscribeNewSideHeads(chan0) + chan1 := make(chan *types.Header) + sub1 := api.events.SubscribeNewSideHeads(chan1) + + go func() { // simulate client + i1, i2 := 0, 0 + for i1 != len(chainSideEvents) || i2 != len(chainSideEvents) { + select { + case header := <-chan0: + if chainSideEvents[i1].Block.Hash() != header.Hash() { + t.Errorf("sub0 received invalid hash on index %d, want %x, got %x", i1, chainSideEvents[i1].Block.Hash(), header.Hash()) + } + i1++ + case header := <-chan1: + if chainSideEvents[i2].Block.Hash() != header.Hash() { + t.Errorf("sub1 received invalid hash on index %d, want %x, got %x", i2, chainSideEvents[i2].Block.Hash(), header.Hash()) + } + i2++ + } + } + + sub0.Unsubscribe() + sub1.Unsubscribe() + }() + + time.Sleep(1 * time.Second) + for _, e := range chainSideEvents { + backend.chainSideFeed.Send(e) + } + + <-sub0.Err() + <-sub1.Err() +} + // TestPendingTxFilter tests whether pending tx filters retrieve all pending transactions that are posted to the event mux. func TestPendingTxFilter(t *testing.T) { t.Parallel() diff --git a/ethclient/ethclient.go b/ethclient/ethclient.go index e8c7c8e7c1..b6d49c2466 100644 --- a/ethclient/ethclient.go +++ b/ethclient/ethclient.go @@ -335,6 +335,12 @@ func (ec *Client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) return ec.c.EthSubscribe(ctx, ch, "newHeads") } +// SubscribeNewSideHead subscribes to notifications about the current blockchain head +// on the given channel. +func (ec *Client) SubscribeNewSideHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { + return ec.c.EthSubscribe(ctx, ch, "newSideHeads") +} + // State Access // NetworkID returns the network ID (also known as the chain ID) for this chain. diff --git a/ethclient/ethclient_test.go b/ethclient/ethclient_test.go index 3c8fe14291..4164a6f9b3 100644 --- a/ethclient/ethclient_test.go +++ b/ethclient/ethclient_test.go @@ -28,6 +28,7 @@ import ( "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/common/math" "github.com/ethereum/go-ethereum/consensus/ethash" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" @@ -418,6 +419,149 @@ func TestRPCDiscover(t *testing.T) { } } +func subscriptionTestSetup(t *testing.T) (genesisBlock *genesisT.Genesis, backend *node.Node) { + // Generate test chain. + // Code largely taken from generateTestChain() + chainConfig := params.TestChainConfig + genesis := &genesisT.Genesis{ + Config: chainConfig, + Alloc: genesisT.GenesisAlloc{testAddr: {Balance: testBalance}}, + ExtraData: []byte("test genesis"), + Timestamp: 9000, + } + + // Create node + // Code largely taken from newTestBackend(t) + backend, err := node.New(&node.Config{}) + if err != nil { + t.Fatalf("can't create new node: %v", err) + } + + // Create Ethereum Service + config := ð.Config{Genesis: genesis} + config.Ethash.PowMode = ethash.ModeFake + + return genesis, backend +} + +func TestEthSubscribeNewSideHeads(t *testing.T) { + + genesis, backend := subscriptionTestSetup(t) + + db := rawdb.NewMemoryDatabase() + chainConfig := genesis.Config + + gblock := core.GenesisToBlock(genesis, db) + engine := ethash.NewFaker() + originalBlocks, _ := core.GenerateChain(chainConfig, gblock, engine, db, 10, func(i int, gen *core.BlockGen) { + gen.OffsetTime(5) + gen.SetExtra([]byte("test")) + }) + originalBlocks = append([]*types.Block{gblock}, originalBlocks...) + + // Create Ethereum Service + config := ð.Config{Genesis: genesis} + config.Ethash.PowMode = ethash.ModeFake + ethservice, err := eth.New(backend, config) + if err != nil { + t.Fatalf("can't create new ethereum service: %v", err) + } + + // Import the test chain. + if err := backend.Start(); err != nil { + t.Fatalf("can't start test node: %v", err) + } + if _, err := ethservice.BlockChain().InsertChain(originalBlocks[1:]); err != nil { + t.Fatalf("can't import test blocks: %v", err) + } + + // Create the client and newSideHeads subscription. + client, err := backend.Attach() + defer backend.Close() + defer client.Close() + if err != nil { + t.Fatal(err) + } + ec := NewClient(client) + defer ec.Close() + + sideHeadCh := make(chan *types.Header) + sub, err := ec.SubscribeNewSideHead(context.Background(), sideHeadCh) + if err != nil { + t.Fatal(err) + } + defer sub.Unsubscribe() + + // Create and import the second-seen chain. + replacementBlocks, _ := core.GenerateChain(chainConfig, originalBlocks[len(originalBlocks)-5], ethservice.Engine(), db, 5, func(i int, gen *core.BlockGen) { + gen.OffsetTime(-9) // difficulty++ + }) + if _, err := ethservice.BlockChain().InsertChain(replacementBlocks); err != nil { + t.Fatalf("can't import test blocks: %v", err) + } + + headersOf := func(bs []*types.Block) (headers []*types.Header) { + for _, b := range bs { + headers = append(headers, b.Header()) + } + return + } + + expectations := []*types.Header{} + + // Why do we expect the replacement (second-seen) blocks reported as side events? + // Because they'll be inserted in ascending order, and until their segment exceeds the total difficulty + // of the incumbent chain, they won't achieve canonical status, despite having greater difficulty per block + // (see the time offset in the block generator function above). + expectations = append(expectations, headersOf(replacementBlocks[:3])...) + + // Once the replacement blocks exceed the total difficulty of the original chain, the + // blocks they replace will be reported as side chain events. + expectations = append(expectations, headersOf(originalBlocks[7:])...) + + // This is illustrated in the logs called below. + for i, b := range originalBlocks { + t.Log("incumbent", i, b.NumberU64(), b.Hash().Hex()[:8]) + } + for i, b := range replacementBlocks { + t.Log("replacement", i, b.NumberU64(), b.Hash().Hex()[:8]) + } + + const timeoutDura = 5 * time.Second + timeout := time.NewTimer(timeoutDura) + + got := []*types.Header{} +waiting: + for { + select { + case head := <-sideHeadCh: + t.Log("<-newSideHeads", head.Number.Uint64(), head.Hash().Hex()[:8]) + got = append(got, head) + if len(got) == len(expectations) { + timeout.Stop() + break waiting + } + timeout.Reset(timeoutDura) + case err := <-sub.Err(): + t.Fatal(err) + case <-timeout.C: + t.Fatal("timed out") + } + } + for i, b := range expectations { + if got[i] == nil { + t.Error("missing expected header (test will improvise a fake value)") + // Set a nonzero value so I don't have to refactor this... + got[i] = &types.Header{Number: big.NewInt(math.MaxInt64)} + } + if got[i].Number.Uint64() != b.Number.Uint64() { + t.Errorf("number: want: %d, got: %d", b.Number.Uint64(), got[i].Number.Uint64()) + } else if got[i].Hash() != b.Hash() { + t.Errorf("hash: want: %s, got: %s", b.Hash().Hex()[:8], got[i].Hash().Hex()[:8]) + } + } +} + // mustNewTestBackend is the same logic as newTestBackend(t *testing.T) but without the testing.T argument. // This function is used exclusively for the benchmarking tests, and will panic if it encounters an error. func mustNewTestBackend() (*node.Node, []*types.Block) { diff --git a/node/openrpc.go b/node/openrpc.go index 1fe0dc781a..33bbff2b07 100644 --- a/node/openrpc.go +++ b/node/openrpc.go @@ -312,6 +312,7 @@ var blockNumberOrHashD = fmt.Sprintf(`{ var rpcSubscriptionParamsNameD = `{ "oneOf": [ {"type": "string", "enum": ["newHeads"], "description": "Fires a notification each time a new header is appended to the chain, including chain reorganizations."}, + {"type": "string", "enum": ["newSideHeads"], "description": "Fires a notification each time a new header is appended to the non-canonical (side) chain, including chain reorganizations."}, {"type": "string", "enum": ["logs"], "description": "Returns logs that are included in new imported blocks and match the given filter criteria."}, {"type": "string", "enum": ["newPendingTransactions"], "description": "Returns the hash for all transactions that are added to the pending state and are signed with a key that is available in the node."}, {"type": "string", "enum": ["syncing"], "description": "Indicates when the node starts or stops synchronizing. The result can either be a boolean indicating that the synchronization has started (true), finished (false) or an object with various progress indicators."} From bf5bd1290fdb0ad59e8d4255100934d053be6e5d Mon Sep 17 00:00:00 2001 From: meows Date: Tue, 12 Jan 2021 08:05:16 -0600 Subject: [PATCH 2/4] filters: (lint) goimports Date: 2021-01-12 08:05:16-06:00 Signed-off-by: meows --- eth/filters/api.go | 1 - 1 file changed, 1 deletion(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index 0221bc962c..97989d86be 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -236,7 +236,6 @@ func (api *PublicFilterAPI) NewSideBlockFilter() rpc.ID { return headerSub.ID } - // NewHeads send a notification each time a new (header) block is appended to the chain. func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) From c8cf61df8b9014620508d71f745b316569fec5e1 Mon Sep 17 00:00:00 2001 From: meows Date: Tue, 12 Jan 2021 09:54:37 -0600 Subject: [PATCH 3/4] filters: (lint) fix comment Resolves https://github.com/etclabscore/core-geth/pull/293/files/bf5bd1290fdb0ad59e8d4255100934d053be6e5d#r555805816 Date: 2021-01-12 09:54:37-06:00 Signed-off-by: meows --- eth/filters/api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index 97989d86be..4e0ab1bbcd 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -266,7 +266,7 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er return rpcSub, nil } -// NewHeads send a notification each time a new (header) block is appended to the chain. +// NewSideHeads send a notification each time a new non-canonical (header) block is written to the database. func (api *PublicFilterAPI) NewSideHeads(ctx context.Context) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { From 85e4d0b644e4ec692d199899dd6e4e34c41ff36c Mon Sep 17 00:00:00 2001 From: meows Date: Tue, 19 Jan 2021 07:07:02 -0600 Subject: [PATCH 4/4] ethclient: install missing eth_newSideBlockFilter method for rpc.discover test Date: 2021-01-19 07:07:02-06:00 Signed-off-by: meows --- ethclient/ethclient_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/ethclient/ethclient_test.go b/ethclient/ethclient_test.go index 4164a6f9b3..89afd906b2 100644 --- a/ethclient/ethclient_test.go +++ b/ethclient/ethclient_test.go @@ -725,6 +725,7 @@ var allRPCMethods = []string{ "eth_hashrate", "eth_mining", "eth_newBlockFilter", + "eth_newSideBlockFilter", "eth_newFilter", "eth_newPendingTransactionFilter", "eth_pendingTransactions",