feat: make ChainHead api local

Author: hunjixin

api/api.go

@@ -56,9 +56,6 @@ type Proxy interface {
 	// between the two objects.
 	ChainStatObj(ctx context.Context, obj cid.Cid, base cid.Cid) (api.ObjStat, error) //perm:read
 
-	// ChainHead returns the current head of the chain.
-	ChainHead(context.Context) (*types.TipSet, error) //perm:read
-
 	// ChainGetBlock returns the block specified by the given CID.
 	ChainGetBlock(context.Context, cid.Cid) (*types.BlockHeader, error) //perm:read
 
@@ -385,6 +382,7 @@ type Local interface {
 	// ChainNotify returns channel with chain head updates.
 	// First message is guaranteed to be of len == 1, and type == 'current'.
 	ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
+	ChainHead(context.Context) (*types.TipSet, error)
 }
 
 // UnSupport is a subset of api.FullNode

co/chain_api.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/filecoin-project/lotus/chain/types"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/store"
 )
@@ -73,3 +75,11 @@ func (c *Coordinator) ChainNotify(ctx context.Context) (<-chan []*api.HeadChange
 
 	return out, nil
 }
+
+// ChainHead impls api.FullNode.ChainNotify
+func (c *Coordinator) ChainHead(in0 context.Context) (*types.TipSet, error) {
+	c.headMu.RLock()
+	defer c.headMu.RUnlock()
+
+	return c.head, nil
+}

co/coordinator.go

@@ -96,13 +96,6 @@ func (c *Coordinator) handleCandidate(hc *headCandidate) {
 
 		prev := c.head
 		next := hc.ts
-		headChanges, err := c.applyTipSetChange(prev, next, hc.node) // todo if network become slow
-		if err != nil {
-			clog.Errorf("apply tipset change: %s", err)
-		}
-		if headChanges == nil {
-			return
-		}
 
 		c.head = hc.ts
 		c.weight = hc.weight
@@ -112,8 +105,16 @@ func (c *Coordinator) handleCandidate(hc *headCandidate) {
 		c.sel.setPriority(DelayPriority, preAddrs...)
 		c.sel.setPriority(CatchUpPriority, addr)
 
-		c.tspub.Pub(headChanges, tipsetChangeTopic)
+		// reorg must after cache update
+		headChanges, err := c.applyTipSetChange(prev, next, hc.node) // todo if network become slow
+		if err != nil {
+			clog.Errorf("apply tipset change: %s", err)
+		}
+		if headChanges == nil {
+			return
+		}
 
+		c.tspub.Pub(headChanges, tipsetChangeTopic)
 		return
 	}
 

co/ctx.go

@@ -14,14 +14,8 @@ import (
 
 // NewCtx constructs a Ctx instance
 func NewCtx(mctx helpers.MetricsCtx, lc fx.Lifecycle, nodeOpt NodeOption) (*Ctx, error) {
-	bcache, err := newBlockHeaderCache(1 << 20)
-	if err != nil {
-		return nil, err
-	}
-
 	return &Ctx{
 		lc:        helpers.LifecycleCtx(mctx, lc),
-		bcache:    bcache,
 		headCh:    make(chan *headCandidate, 256),
 		errNodeCh: make(chan string, 256),
 		nodeOpt:   nodeOpt,
@@ -31,7 +25,6 @@ func NewCtx(mctx helpers.MetricsCtx, lc fx.Lifecycle, nodeOpt NodeOption) (*Ctx,
 // Ctx contains the shared components between different modules
 type Ctx struct {
 	lc        context.Context
-	bcache    *blockHeaderCache
 	headCh    chan *headCandidate
 	errNodeCh chan string
 
@@ -77,3 +70,18 @@ func (bc *blockHeaderCache) load(c cid.Cid) (*types.BlockHeader, bool) {
 	blk, ok := val.(*types.BlockHeader)
 	return blk, ok
 }
+
+func (bc *blockHeaderCache) has(c cid.Cid) bool {
+	_, ok := bc.cache.Peek(c)
+	return ok
+}
+
+func (bc *blockHeaderCache) hasKey(key types.TipSetKey) bool {
+	for _, blkCid := range key.Cids() {
+		has := bc.has(blkCid)
+		if !has {
+			return false
+		}
+	}
+	return true
+}

co/node.go

@@ -5,8 +5,6 @@ import (
 	"sync"
 	"time"
 
-	lru "github.com/hashicorp/golang-lru"
-
 	"github.com/filecoin-project/lotus/api/v1api"
 	"github.com/ipfs-force-community/venus-common-utils/apiinfo"
 
@@ -73,8 +71,9 @@ type Node struct {
 		closer jsonrpc.ClientCloser
 	}
 
-	blkCache *lru.ARCCache
-	log      *zap.SugaredLogger
+	blkCache *blockHeaderCache
+
+	log *zap.SugaredLogger
 }
 
 func NewNode(cctx *Ctx, info NodeInfo) (*Node, error) {
@@ -83,11 +82,12 @@ func NewNode(cctx *Ctx, info NodeInfo) (*Node, error) {
 		return nil, err
 	}
 	ctx, cancel := context.WithCancel(cctx.lc)
-	blkCache, err := lru.NewARC(100)
+	blkCache, err := newBlockHeaderCache(1 << 20)
 	if err != nil {
 		cancel()
 		return nil, err
 	}
+
 	return &Node{
 		reListenInterval: cctx.nodeOpt.ReListenMinInterval,
 		opt:              cctx.nodeOpt,
@@ -211,7 +211,7 @@ func (n *Node) reListen() (<-chan []*api.HeadChange, error) {
 }
 
 func (n *Node) applyChanges(lifeCtx context.Context, changes []*api.HeadChange) {
-	n.sctx.bcache.add(changes)
+	n.blkCache.add(changes)
 
 	idx := -1
 	for i := range changes {
@@ -220,7 +220,6 @@ func (n *Node) applyChanges(lifeCtx context.Context, changes []*api.HeadChange)
 			idx = i
 		case store.HCApply:
 			idx = i
-			n.storeKey(changes[i].Val.Key())
 		}
 	}
 
@@ -292,31 +291,16 @@ func (n *Node) loadTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipS
 }
 
 func (n *Node) loadBlockHeader(ctx context.Context, c cid.Cid) (*types.BlockHeader, error) {
-	if blk, ok := n.sctx.bcache.load(c); ok {
+	if blk, ok := n.blkCache.load(c); ok {
 		return blk, nil
 	}
 
 	blk, err := n.upstream.full.ChainGetBlock(ctx, c)
 	return blk, err
 }
 
-func (n *Node) hasKey(key types.TipSetKey) bool {
-	for _, blkCid := range key.Cids() {
-		_, has := n.blkCache.Peek(blkCid.String())
-		if !has {
-			return false
-		}
-	}
-	return true
-}
-
-func (n *Node) storeKey(key types.TipSetKey) {
-	for _, blkCid := range key.Cids() {
-		_, has := n.blkCache.Peek(blkCid.String())
-		if !has {
-			n.blkCache.Add(blkCid.String(), nil)
-		}
-	}
+func (n *Node) hasTipset(key types.TipSetKey) bool {
+	return n.blkCache.hasKey(key)
 }
 
 const (

co/selector.go

@@ -157,8 +157,9 @@ func (s *Selector) Select(tsk types.TipSetKey) (*Node, error) {
 
 	for addr, p := range s.priority {
 		node := s.nodeProvider.GetNode(addr)
-		if !node.hasKey(tsk) {
-			log.Warnf("node %s not contains key %s", addr, tsk)
+		if !tsk.IsEmpty() && node.hasTipset(tsk) && p != ErrPriority {
+			log.Debugf("node %s has tipset %s, change to catchup node", addr, tsk.Cids())
+			catchUpQue[addr] = s.weight[addr]
 			continue
 		}
 		if p == CatchUpPriority {

proxy/local.go

@@ -20,6 +20,15 @@ type Local struct {
 }
 
 // impl api.Local
+func (p *Local) ChainHead(in0 context.Context) (out0 *types.TipSet, err error) {
+	cli, err := p.Select(types.EmptyTSK)
+	if err != nil {
+		err = fmt.Errorf("api ChainHead %v", err)
+		return
+	}
+	return cli.ChainHead(in0)
+}
+
 func (p *Local) ChainNotify(in0 context.Context) (out0 <-chan []*api1.HeadChange, err error) {
 	cli, err := p.Select(types.EmptyTSK)
 	if err != nil {

proxy/proxy.go

@@ -140,15 +140,6 @@ func (p *Proxy) ChainHasObj(in0 context.Context, in1 cid.Cid) (out0 bool, err er
 	return cli.ChainHasObj(in0, in1)
 }
 
-func (p *Proxy) ChainHead(in0 context.Context) (out0 *types.TipSet, err error) {
-	cli, err := p.Select(types.EmptyTSK)
-	if err != nil {
-		err = fmt.Errorf("api ChainHead %v", err)
-		return
-	}
-	return cli.ChainHead(in0)
-}
-
 func (p *Proxy) ChainReadObj(in0 context.Context, in1 cid.Cid) (out0 []uint8, err error) {
 	cli, err := p.Select(types.EmptyTSK)
 	if err != nil {