Merge pull request #1913 from acruikshank/feat/1853-confirmations-in-StateWaitMsg2

Add confidence and timeout to message wait

api/api_full.go

@@ -156,7 +156,7 @@ type FullNode interface {
 	StateSectorPreCommitInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error)
 	StateSectorGetInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (*miner.SectorOnChainInfo, error)
 	StatePledgeCollateral(context.Context, types.TipSetKey) (types.BigInt, error)
-	StateWaitMsg(context.Context, cid.Cid) (*MsgLookup, error)
+	StateWaitMsg(ctx context.Context, cid cid.Cid, confidence uint64) (*MsgLookup, error)
 	StateSearchMsg(context.Context, cid.Cid) (*MsgLookup, error)
 	StateListMiners(context.Context, types.TipSetKey) ([]address.Address, error)
 	StateListActors(context.Context, types.TipSetKey) ([]address.Address, error)

api/apistruct/struct.go

@@ -136,7 +136,7 @@ type FullNodeStruct struct {
 		StateGetActor                     func(context.Context, address.Address, types.TipSetKey) (*types.Actor, error)                                       `perm:"read"`
 		StateReadState                    func(context.Context, *types.Actor, types.TipSetKey) (*api.ActorState, error)                                       `perm:"read"`
 		StatePledgeCollateral             func(context.Context, types.TipSetKey) (types.BigInt, error)                                                        `perm:"read"`
-		StateWaitMsg                      func(context.Context, cid.Cid) (*api.MsgLookup, error)                                                              `perm:"read"`
+		StateWaitMsg                      func(ctx context.Context, cid cid.Cid, confidence uint64) (*api.MsgLookup, error)                                   `perm:"read"`
 		StateSearchMsg                    func(context.Context, cid.Cid) (*api.MsgLookup, error)                                                              `perm:"read"`
 		StateListMiners                   func(context.Context, types.TipSetKey) ([]address.Address, error)                                                   `perm:"read"`
 		StateListActors                   func(context.Context, types.TipSetKey) ([]address.Address, error)                                                   `perm:"read"`
@@ -594,8 +594,8 @@ func (c *FullNodeStruct) StatePledgeCollateral(ctx context.Context, tsk types.Ti
 	return c.Internal.StatePledgeCollateral(ctx, tsk)
 }
 
-func (c *FullNodeStruct) StateWaitMsg(ctx context.Context, msgc cid.Cid) (*api.MsgLookup, error) {
+func (c *FullNodeStruct) StateWaitMsg(ctx context.Context, msgc cid.Cid, confidence uint64) (*api.MsgLookup, error) {
-	return c.Internal.StateWaitMsg(ctx, msgc)
+	return c.Internal.StateWaitMsg(ctx, msgc, confidence)
 }
 
 func (c *FullNodeStruct) StateSearchMsg(ctx context.Context, msgc cid.Cid) (*api.MsgLookup, error) {

build/params_shared.go

@@ -59,6 +59,7 @@ var BlocksPerEpoch = uint64(builtin.ExpectedLeadersPerEpoch)
 
 // Epochs
 const Finality = miner.ChainFinalityish
+const MessageConfidence = 5
 
 // constants for Weight calculation
 // The ratio of weight contributed by short-term vs long-term factors in a given round

chain/stmgr/stmgr.go

@@ -480,7 +480,10 @@ func (sm *StateManager) GetReceipt(ctx context.Context, msg cid.Cid, ts *types.T
 	return r, nil
 }
 
-func (sm *StateManager) WaitForMessage(ctx context.Context, mcid cid.Cid) (*types.TipSet, *types.MessageReceipt, error) {
+// WaitForMessage blocks until a message appears on chain. It looks backwards in the chain to see if this has already
+// happened. It guarantees that the message has been on chain for at least confidence epochs without being reverted
+// before returning.
+func (sm *StateManager) WaitForMessage(ctx context.Context, mcid cid.Cid, confidence uint64) (*types.TipSet, *types.MessageReceipt, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -528,6 +531,11 @@ func (sm *StateManager) WaitForMessage(ctx context.Context, mcid cid.Cid) (*type
 		close(backSearchWait)
 	}()
 
+	var candidateTs *types.TipSet
+	var candidateRcp *types.MessageReceipt
+	heightOfHead := head[0].Val.Height()
+	reverts := map[types.TipSetKey]bool{}
+
 	for {
 		select {
 		case notif, ok := <-tsub:
@@ -537,21 +545,44 @@ func (sm *StateManager) WaitForMessage(ctx context.Context, mcid cid.Cid) (*type
 			for _, val := range notif {
 				switch val.Type {
 				case store.HCRevert:
-					continue
+					if val.Val.Equals(candidateTs) {
+						candidateTs = nil
+						candidateRcp = nil
+					}
+					if backSearchWait != nil {
+						reverts[val.Val.Key()] = true
+					}
 				case store.HCApply:
+					if candidateTs != nil && val.Val.Height() >= candidateTs.Height()+abi.ChainEpoch(confidence) {
+						return candidateTs, candidateRcp, nil
+					}
 					r, err := sm.tipsetExecutedMessage(val.Val, mcid, msg.VMMessage())
 					if err != nil {
 						return nil, nil, err
 					}
 					if r != nil {
-						return val.Val, r, nil
+						if confidence == 0 {
+							return val.Val, r, err
+						}
+						candidateTs = val.Val
+						candidateRcp = r
 					}
+					heightOfHead = val.Val.Height()
 				}
 			}
 		case <-backSearchWait:
-			if backTs != nil {
+			// check if we found the message in the chain and that is hasn't been reverted since we started searching
-				return backTs, backRcp, nil
+			if backTs != nil && !reverts[backTs.Key()] {
+				// if head is at or past confidence interval, return immediately
+				if heightOfHead >= backTs.Height()+abi.ChainEpoch(confidence) {
+					return backTs, backRcp, nil
+				}
+
+				// wait for confidence interval
+				candidateTs = backTs
+				candidateRcp = backRcp
 			}
+			reverts = nil
 			backSearchWait = nil
 		case <-ctx.Done():
 			return nil, nil, ctx.Err()

cli/multisig.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/apibstore"
+	"github.com/filecoin-project/lotus/build"
 	types "github.com/filecoin-project/lotus/chain/types"
 )
 
@@ -117,7 +118,7 @@ var msigCreateCmd = &cli.Command{
 		}
 
 		// wait for it to get mined into a block
-		wait, err := api.StateWaitMsg(ctx, msgCid)
+		wait, err := api.StateWaitMsg(ctx, msgCid, build.MessageConfidence)
 		if err != nil {
 			return err
 		}
@@ -336,7 +337,7 @@ var msigProposeCmd = &cli.Command{
 
 		fmt.Println("send proposal in message: ", msgCid)
 
-		wait, err := api.StateWaitMsg(ctx, msgCid)
+		wait, err := api.StateWaitMsg(ctx, msgCid, build.MessageConfidence)
 		if err != nil {
 			return err
 		}
@@ -452,7 +453,7 @@ var msigApproveCmd = &cli.Command{
 
 		fmt.Println("sent approval in message: ", msgCid)
 
-		wait, err := api.StateWaitMsg(ctx, msgCid)
+		wait, err := api.StateWaitMsg(ctx, msgCid, build.MessageConfidence)
 		if err != nil {
 			return err
 		}

cli/paych.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/base64"
 	"fmt"
+	"github.com/filecoin-project/lotus/build"
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
@@ -361,7 +362,7 @@ var paychVoucherSubmitCmd = &cli.Command{
 			return err
 		}
 
-		mwait, err := api.StateWaitMsg(ctx, mcid)
+		mwait, err := api.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 		if err != nil {
 			return err
 		}

cli/state.go

@@ -35,6 +35,7 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
 
 	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/miner"
 )
@@ -392,7 +393,7 @@ var stateReplaySetCmd = &cli.Command{
 
 				ts, err = types.NewTipSet(headers)
 			} else {
-				r, err := api.StateWaitMsg(ctx, mcid)
+				r, err := api.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 				if err != nil {
 					return xerrors.Errorf("finding message in chain: %w", err)
 				}
@@ -1156,7 +1157,7 @@ var stateWaitMsgCmd = &cli.Command{
 			return err
 		}
 
-		mw, err := api.StateWaitMsg(ctx, msg)
+		mw, err := api.StateWaitMsg(ctx, msg, build.MessageConfidence)
 		if err != nil {
 			return err
 		}

cmd/lotus-fountain/main.go

@@ -334,7 +334,7 @@ func (h *handler) msgwait(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	mw, err := h.api.StateWaitMsg(r.Context(), c)
+	mw, err := h.api.StateWaitMsg(r.Context(), c, build.MessageConfidence)
 	if err != nil {
 		w.WriteHeader(400)
 		w.Write([]byte(err.Error()))
@@ -357,7 +357,7 @@ func (h *handler) msgwaitaddr(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	mw, err := h.api.StateWaitMsg(r.Context(), c)
+	mw, err := h.api.StateWaitMsg(r.Context(), c, build.MessageConfidence)
 	if err != nil {
 		w.WriteHeader(400)
 		w.Write([]byte(err.Error()))

cmd/lotus-shed/verifreg.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"fmt"
+	"github.com/filecoin-project/lotus/build"
 
 	"github.com/filecoin-project/go-address"
 	"gopkg.in/urfave/cli.v2"
@@ -85,7 +86,7 @@ var verifRegAddVerifierCmd = &cli.Command{
 
 		fmt.Printf("message sent, now waiting on cid: %s\n", smsg.Cid())
 
-		mwait, err := api.StateWaitMsg(ctx, smsg.Cid())
+		mwait, err := api.StateWaitMsg(ctx, smsg.Cid(), build.MessageConfidence)
 		if err != nil {
 			return err
 		}
@@ -161,7 +162,7 @@ var verifRegVerifyClientCmd = &cli.Command{
 
 		fmt.Printf("message sent, now waiting on cid: %s\n", smsg.Cid())
 
-		mwait, err := api.StateWaitMsg(ctx, smsg.Cid())
+		mwait, err := api.StateWaitMsg(ctx, smsg.Cid(), build.MessageConfidence)
 		if err != nil {
 			return err
 		}

cmd/lotus-storage-miner/init.go

@@ -566,7 +566,7 @@ func configureStorageMiner(ctx context.Context, api lapi.FullNode, addr address.
 	}
 
 	log.Info("Waiting for message: ", smsg.Cid())
-	ret, err := api.StateWaitMsg(ctx, smsg.Cid())
+	ret, err := api.StateWaitMsg(ctx, smsg.Cid(), build.MessageConfidence)
 	if err != nil {
 		return err
 	}
@@ -648,7 +648,7 @@ func createStorageMiner(ctx context.Context, api lapi.FullNode, peerid peer.ID,
 	log.Infof("Pushed StorageMarket.CreateStorageMiner, %s to Mpool", signed.Cid())
 	log.Infof("Waiting for confirmation")
 
-	mw, err := api.StateWaitMsg(ctx, signed.Cid())
+	mw, err := api.StateWaitMsg(ctx, signed.Cid(), build.MessageConfidence)
 	if err != nil {
 		return address.Undef, err
 	}

markets/retrievaladapter/client.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	payapi "github.com/filecoin-project/lotus/node/impl/paych"
 	"github.com/filecoin-project/lotus/paychmgr"
@@ -72,7 +73,7 @@ func (rcn *retrievalClientNode) GetChainHead(ctx context.Context) (shared.TipSet
 // WaitForPaymentChannelAddFunds waits messageCID to appear on chain. If it doesn't appear within
 // defaultMsgWaitTimeout it returns error
 func (rcn *retrievalClientNode) WaitForPaymentChannelAddFunds(messageCID cid.Cid) error {
-	_, mr, err := rcn.chainapi.StateManager.WaitForMessage(context.TODO(), messageCID)
+	_, mr, err := rcn.chainapi.StateManager.WaitForMessage(context.TODO(), messageCID, build.MessageConfidence)
 
 	if err != nil {
 		return err
@@ -84,7 +85,7 @@ func (rcn *retrievalClientNode) WaitForPaymentChannelAddFunds(messageCID cid.Cid
 }
 
 func (rcn *retrievalClientNode) WaitForPaymentChannelCreation(messageCID cid.Cid) (address.Address, error) {
-	_, mr, err := rcn.chainapi.StateManager.WaitForMessage(context.TODO(), messageCID)
+	_, mr, err := rcn.chainapi.StateManager.WaitForMessage(context.TODO(), messageCID, build.MessageConfidence)
 
 	if err != nil {
 		return address.Undef, err

markets/storageadapter/client.go

@@ -211,7 +211,7 @@ func (c *ClientNodeAdapter) ValidatePublishedDeal(ctx context.Context, deal stor
 	}
 
 	// TODO: timeout
-	_, ret, err := c.sm.WaitForMessage(ctx, *deal.PublishMessage)
+	_, ret, err := c.sm.WaitForMessage(ctx, *deal.PublishMessage, build.MessageConfidence)
 	if err != nil {
 		return 0, xerrors.Errorf("waiting for deal publish message: %w", err)
 	}
@@ -321,7 +321,7 @@ func (c *ClientNodeAdapter) OnDealSectorCommitted(ctx context.Context, provider
 		}
 	}
 
-	if err := c.ev.Called(checkFunc, called, revert, 3, build.SealRandomnessLookbackLimit, matchEvent); err != nil {
+	if err := c.ev.Called(checkFunc, called, revert, build.MessageConfidence+1, build.SealRandomnessLookbackLimit, matchEvent); err != nil {
 		return xerrors.Errorf("failed to set up called handler: %w", err)
 	}
 
@@ -397,7 +397,7 @@ func (n *ClientNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetToke
 }
 
 func (n *ClientNodeAdapter) WaitForMessage(ctx context.Context, mcid cid.Cid, cb func(code exitcode.ExitCode, bytes []byte, err error) error) error {
-	receipt, err := n.StateWaitMsg(ctx, mcid)
+	receipt, err := n.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 	if err != nil {
 		return cb(0, nil, err)
 	}

markets/storageadapter/provider.go

@@ -321,7 +321,7 @@ func (n *ProviderNodeAdapter) OnDealSectorCommitted(ctx context.Context, provide
 
 	}
 
-	if err := n.ev.Called(checkFunc, called, revert, 3, build.SealRandomnessLookbackLimit, matchEvent); err != nil {
+	if err := n.ev.Called(checkFunc, called, revert, build.MessageConfidence+1, build.SealRandomnessLookbackLimit, matchEvent); err != nil {
 		return xerrors.Errorf("failed to set up called handler: %w", err)
 	}
 
@@ -338,7 +338,7 @@ func (n *ProviderNodeAdapter) GetChainHead(ctx context.Context) (shared.TipSetTo
 }
 
 func (n *ProviderNodeAdapter) WaitForMessage(ctx context.Context, mcid cid.Cid, cb func(code exitcode.ExitCode, bytes []byte, err error) error) error {
-	receipt, err := n.StateWaitMsg(ctx, mcid)
+	receipt, err := n.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 	if err != nil {
 		return cb(0, nil, err)
 	}

node/impl/full/state.go

@@ -344,10 +344,8 @@ func (a *StateAPI) MinerCreateBlock(ctx context.Context, bt *api.BlockTemplate)
 	return &out, nil
 }
 
-func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
+func (a *StateAPI) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error) {
-	// TODO: consider using event system for this, expose confidence
+	ts, recpt, err := a.StateManager.WaitForMessage(ctx, msg, confidence)
-
-	ts, recpt, err := a.StateManager.WaitForMessage(ctx, msg)
 	if err != nil {
 		return nil, err
 	}

paychmgr/simple.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 
+	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/types"
 )
@@ -55,7 +56,7 @@ func (pm *Manager) createPaych(ctx context.Context, from, to address.Address, am
 //  (tricky because we need to setup channel tracking before we know its address)
 func (pm *Manager) waitForPaychCreateMsg(ctx context.Context, mcid cid.Cid) {
 	defer pm.store.lk.Unlock()
-	mwait, err := pm.state.StateWaitMsg(ctx, mcid)
+	mwait, err := pm.state.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 	if err != nil {
 		log.Errorf("wait msg: %w", err)
 	}
@@ -105,7 +106,7 @@ func (pm *Manager) addFunds(ctx context.Context, ch address.Address, from addres
 //  (tricky because we need to setup channel tracking before we know it's address)
 func (pm *Manager) waitForAddFundsMsg(ctx context.Context, mcid cid.Cid) {
 	defer pm.store.lk.Unlock()
-	mwait, err := pm.state.StateWaitMsg(ctx, mcid)
+	mwait, err := pm.state.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 	if err != nil {
 		log.Error(err)
 	}

storage/adapter_storage_miner.go

@@ -18,6 +18,7 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
 
 	"github.com/filecoin-project/lotus/api/apibstore"
+	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/actors"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -81,7 +82,7 @@ func (s SealingAPIAdapter) StateMinerDeadlines(ctx context.Context, maddr addres
 }
 
 func (s SealingAPIAdapter) StateWaitMsg(ctx context.Context, mcid cid.Cid) (sealing.MsgLookup, error) {
-	wmsg, err := s.delegate.StateWaitMsg(ctx, mcid)
+	wmsg, err := s.delegate.StateWaitMsg(ctx, mcid, build.MessageConfidence)
 	if err != nil {
 		return sealing.MsgLookup{}, err
 	}

storage/miner.go

@@ -54,7 +54,7 @@ type storageMinerApi interface {
 	StateMinerInfo(context.Context, address.Address, types.TipSetKey) (miner.MinerInfo, error)
 	StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*miner.DeadlineInfo, error)
 	StateMinerInitialPledgeCollateral(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (types.BigInt, error)
-	StateWaitMsg(context.Context, cid.Cid) (*api.MsgLookup, error) // TODO: removeme eventually
+	StateWaitMsg(ctx context.Context, cid cid.Cid, confidence uint64) (*api.MsgLookup, error) // TODO: removeme eventually
 	StateGetActor(ctx context.Context, actor address.Address, ts types.TipSetKey) (*types.Actor, error)
 	StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
 	StateMarketStorageDeal(context.Context, abi.DealID, types.TipSetKey) (*api.MarketDeal, error)

storage/wdpost_run.go

@@ -255,7 +255,7 @@ func (s *WindowPoStScheduler) checkNextFaults(ctx context.Context, deadline uint
 
 	log.Warnw("declare faults Message CID", "cid", sm.Cid())
 
-	rec, err := s.api.StateWaitMsg(context.TODO(), sm.Cid())
+	rec, err := s.api.StateWaitMsg(context.TODO(), sm.Cid(), build.MessageConfidence)
 	if err != nil {
 		return xerrors.Errorf("declare faults wait error: %w", err)
 	}
@@ -499,7 +499,7 @@ func (s *WindowPoStScheduler) submitPost(ctx context.Context, proof *miner.Submi
 	log.Infof("Submitted window post: %s", sm.Cid())
 
 	go func() {
-		rec, err := s.api.StateWaitMsg(context.TODO(), sm.Cid())
+		rec, err := s.api.StateWaitMsg(context.TODO(), sm.Cid(), build.MessageConfidence)
 		if err != nil {
 			log.Error(err)
 			return