Pandora can send previous block information while resubscribed (#112)

- Fix pending block subscription api Co-authored-by: meta-bot <[email protected]> Co-authored-by: Atif Anowar <[email protected]>
frozeman · Nov 26, 2021 · e6b5eef · e6b5eef
1 parent 2addfe8
commit e6b5eef
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 13 deletions.
diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go
@@ -836,6 +836,10 @@ func (fb *filterBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*t
 	return fb.bc.GetHeaderByHash(hash), nil
 }
 
+func (fb *filterBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error) {
+	return nil, nil
+}
+
 func (fb *filterBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
 	number := rawdb.ReadHeaderNumber(fb.db, hash)
 	if number == nil {

diff --git a/eth/filters/api.go b/eth/filters/api.go
@@ -228,15 +228,33 @@ func (api *PublicFilterAPI) NewPendingBlockHeaders(ctx context.Context, pendingF
 	go func() {
 		// sending previous blocks to orchestrator
 		sender := func(windowStart, windowEnd uint64) {
+			log.Debug("sending header to orchestrator", "windowStart", windowStart, "windowEnd", windowEnd)
 			for i := windowStart; i <= windowEnd; i++ {
-				temphead, err := api.backend.HeaderByNumber(ctx, rpc.BlockNumber(i))
+				temphead, err := api.backend.BlockByNumber(ctx, rpc.BlockNumber(i))
 				if temphead != nil && err == nil {
 					// if block exists and no error occurred then send it
-					log.Debug("Notifying to orchestrator", "block number", temphead.Number.Uint64())
-					notifier.Notify(rpcSub.ID, temphead)
+					log.Debug("Notifying to orchestrator", "block number", temphead.NumberU64())
+					notifier.Notify(rpcSub.ID, temphead.Header())
+				} else {
+					log.Debug("sending failed", "consideredBlockNumber", i, "tempHead", temphead, "err", err)
 				}
 			}
 		}
+		sendByHash := func(start uint64, blockHash common.Hash) {
+			log.Info("sending header to orchestrator", "start", start, "blockHash", blockHash)
+			var headerArray []*types.Header
+			for header, _ := api.backend.HeaderByHash(ctx, blockHash); header != nil; header, _ = api.backend.HeaderByHash(ctx, header.ParentHash) {
+				headerArray = append(headerArray, header)
+				if header.Number.Uint64() == start {
+					break
+				}
+			}
+			log.Debug("sendByHash found headers of length", "len", len(headerArray))
+			for i := len(headerArray) - 1; i >= 0; i-- {
+				log.Debug("Notifying to orchestrator", "block number", headerArray[i].Number.Uint64())
+				notifier.Notify(rpcSub.ID, headerArray[i])
+			}
+		}
 		// we are starting from block 1 because block 0 has no pandora extra data info. so sending it will create an error
 		windowStart, windowEnd := uint64(1), uint64(1)
 		header, _ := api.backend.HeaderByHash(ctx, pendingFilter.FromBlockHash)
@@ -273,22 +291,20 @@ func (api *PublicFilterAPI) NewPendingBlockHeaders(ctx context.Context, pendingF
 		// If anything goes wrong remove it.
 		headers := make(chan *types.Header)
 		headersSub := api.events.SubscribePendingHeads(headers)
-		firstTime := true
 
 		for {
 			select {
 			case h := <-headers:
-				if firstTime {
-					// entered into the running phase. for the first time we will do checking and send previous blocks.
-					firstTime = false
-					if windowEnd+1 < h.Number.Uint64() {
-						// not consecutive. so send them first
-						windowStart = windowEnd + 1
-						windowEnd = h.Number.Uint64()
-						sender(windowStart, windowEnd)
-					}
+				if h != nil && windowEnd+1 < h.Number.Uint64() {
+					windowStart = windowEnd + 1
+					windowEnd = h.Number.Uint64() - 1
+					log.Debug("previous block information are not sent. so resending", "windowStart", windowStart, "windowEnd", windowEnd)
+					sendByHash(windowStart, h.ParentHash)
 				}
 				notifier.Notify(rpcSub.ID, h)
+				if h != nil {
+					windowEnd = h.Number.Uint64()
+				}
 			case rpcErr := <-rpcSub.Err():
 				log.Debug("error found in rpc subscription", "error", rpcErr)
 				headersSub.Unsubscribe()

diff --git a/eth/filters/filter.go b/eth/filters/filter.go
@@ -34,6 +34,7 @@ type Backend interface {
 	ChainDb() ethdb.Database
 	HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error)
 	HeaderByHash(ctx context.Context, blockHash common.Hash) (*types.Header, error)
+	BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error)
 	GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
 	GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
 	GetPendingHeadsSince(ctx context.Context, from common.Hash) []*types.Header

diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go
@@ -86,6 +86,10 @@ func (b *testBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*type
 	return rawdb.ReadHeader(b.db, hash, *number), nil
 }
 
+func (b *testBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error) {
+	return nil, nil
+}
+
 func (b *testBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
 	if number := rawdb.ReadHeaderNumber(b.db, hash); number != nil {
 		return rawdb.ReadReceipts(b.db, hash, *number, params.TestChainConfig), nil