bugfix: added another data path for block request in the case of regi…

…on doesn't get the broadcast

Added another data path if dom doesnt receive a broadcast, sub will inform them

core/core.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"context"
 	"errors"
 	"io"
 	"math/big"
@@ -193,6 +194,17 @@ func (c *Core) serviceBlocks(hashNumberList []types.HashAndNumber) {
 			}
 			parentHeader := c.GetHeader(header.ParentHash(), header.NumberU64()-1)
 			if parentHeader != nil {
+				// If parent header is dom, send a signal to dom to request for the block if it doesnt have it
+				_, parentHeaderOrder, err := c.sl.engine.CalcOrder(parentHeader)
+				if err != nil {
+					log.Warn("Error calculating the parent block order in serviceBlocks", "Hash", parentHeader.Hash(), "Number", parentHeader.NumberArray())
+				}
+				nodeCtx := common.NodeLocation.Context()
+				if parentHeaderOrder < nodeCtx && c.GetHeaderByHash(parentHeader.Hash()) == nil {
+					log.Info("Requesting the dom to get the block if it doesnt have and try to append", "Hash", parentHeader.Hash(), "Order", parentHeaderOrder)
+					// send a signal to the required dom to fetch the block if it doesnt have it, or its not in its appendqueue
+					go c.sl.domClient.RequestDomToAppendOrFetch(context.Background(), parentHeader.Hash(), parentHeaderOrder)
+				}
 				c.InsertChain([]*types.Block{types.NewBlockWithHeader(header)})
 			} else {
 				if !c.HasHeader(header.ParentHash(), header.NumberU64()-1) {
@@ -205,6 +217,30 @@ func (c *Core) serviceBlocks(hashNumberList []types.HashAndNumber) {
 	}
 }
 
+func (c *Core) RequestDomToAppendOrFetch(hash common.Hash, order int) {
+	// TODO: optimize to check if the block is in the appendqueue or already
+	// appended to reduce the network bandwidth utilization
+	nodeCtx := common.NodeLocation.Context()
+	if nodeCtx == common.PRIME_CTX {
+		// If prime all you can do it to ask for the block
+		_, exists := c.appendQueue.Get(hash)
+		if !exists {
+			log.Warn("Block sub asked doesnt exist in append queue, so request the peers for it", "Hash", hash, "Order", order)
+			c.sl.missingParentFeed.Send(hash)
+		}
+	} else if nodeCtx == common.REGION_CTX {
+		if order < nodeCtx { // Prime block
+			go c.sl.domClient.RequestDomToAppendOrFetch(context.Background(), hash, order)
+		}
+		_, exists := c.appendQueue.Get(hash)
+		if !exists {
+			log.Warn("Block sub asked doesnt exist in append queue, so request the peers for it", "Hash", hash, "Order", order)
+			c.sl.missingParentFeed.Send(hash)
+		}
+	}
+
+}
+
 // addToAppendQueue adds a block to the append queue
 func (c *Core) addToAppendQueue(block *types.Block) error {
 	c.appendQueue.ContainsOrAdd(block.Hash(), blockNumberAndRetryCounter{block.NumberU64(), 0})
@@ -306,8 +342,12 @@ func (c *Core) WriteBlock(block *types.Block) {
 		if err != nil {
 			return
 		}
-		if order == common.NodeLocation.Context() {
+		nodeCtx := common.NodeLocation.Context()
+		if order == nodeCtx {
 			c.addToAppendQueue(block)
+			// If a dom block comes in and we havent appended it yet
+		} else if order < nodeCtx && c.GetHeaderByHash(block.Hash()) == nil {
+			go c.sl.domClient.RequestDomToAppendOrFetch(context.Background(), block.Hash(), order)
 		}
 	}
 	if c.GetHeaderOrCandidateByHash(block.Hash()) == nil {

core/slice.go

@@ -339,6 +339,8 @@ func (sl *Slice) relayPh(block *types.Block, pendingHeaderWithTermini types.Pend
 			bestPh.Header().SetLocation(common.NodeLocation)
 			sl.miner.worker.pendingHeaderFeed.Send(bestPh.Header())
 			return
+		} else {
+			log.Warn("Pending Header for Best ph key does not exist", "best ph key", sl.bestPhKey)
 		}
 	} else if !domOrigin {
 		for _, i := range sl.randomRelayArray() {

eth/api_backend.go

@@ -476,6 +476,10 @@ func (b *QuaiAPIBackend) UpdateDom(oldTerminus common.Hash, newTerminus common.H
 	b.eth.core.UpdateDom(oldTerminus, newTerminus, location)
 }
 
+func (b *QuaiAPIBackend) RequestDomToAppendOrFetch(hash common.Hash, order int) {
+	b.eth.core.RequestDomToAppendOrFetch(hash, order)
+}
+
 func (b *QuaiAPIBackend) ProcessingState() bool {
 	return b.eth.core.ProcessingState()
 }

internal/quaiapi/backend.go

@@ -78,6 +78,7 @@ type Backend interface {
 	PendingBlock() *types.Block
 	SubRelayPendingHeader(pendingHeader types.PendingHeader, location common.Location, subReorg bool)
 	UpdateDom(oldTerminus common.Hash, newTerminus common.Hash, location common.Location)
+	RequestDomToAppendOrFetch(hash common.Hash, order int)
 	NewGenesisPendingHeader(pendingHeader *types.Header)
 	GetPendingHeader() (*types.Header, error)
 	GetManifest(blockHash common.Hash) (types.BlockManifest, error)

internal/quaiapi/quai_api.go

@@ -658,6 +658,18 @@ func (s *PublicBlockChainQuaiAPI) UpdateDom(ctx context.Context, raw json.RawMes
 	s.b.UpdateDom(domUpdate.OldTerminus, domUpdate.NewTerminus, domUpdate.Location)
 }
 
+type RequestDomArgs struct {
+	Hash  common.Hash
+	Order int
+}
+
+func (s *PublicBlockChainQuaiAPI) RequestDomToAppendOrFetch(ctx context.Context, raw json.RawMessage) {
+	var requestDom RequestDomArgs
+	if err := json.Unmarshal(raw, &requestDom); err != nil {
+		return
+	}
+	s.b.RequestDomToAppendOrFetch(requestDom.Hash, requestDom.Order)
+}
 func (s *PublicBlockChainQuaiAPI) NewGenesisPendingHeader(ctx context.Context, raw json.RawMessage) {
 	var pendingHeader *types.Header
 	if err := json.Unmarshal(raw, &pendingHeader); err != nil {

quaiclient/quaiclient.go

@@ -128,6 +128,13 @@ func (ec *Client) UpdateDom(ctx context.Context, oldTerminus common.Hash, newTer
 	ec.c.CallContext(ctx, nil, "quai_updateDom", data)
 }
 
+func (ec *Client) RequestDomToAppendOrFetch(ctx context.Context, hash common.Hash, order int) {
+	data := map[string]interface{}{"Hash": hash}
+	data["Order"] = order
+
+	ec.c.CallContext(ctx, nil, "quai_requestDomToAppendOrFetch", data)
+}
+
 func (ec *Client) NewGenesisPendingHeader(ctx context.Context, header *types.Header) {
 	ec.c.CallContext(ctx, nil, "quai_newGenesisPendingHeader", header.RPCMarshalHeader())
 }