Merge pull request #2958 from filecoin-project/asr/sync-test

MessagesForTipset should filter out bad nonce messages

chain/gen/gen.go

@@ -272,6 +272,10 @@ func NewGenerator() (*ChainGen, error) {
 	return NewGeneratorWithSectors(1)
 }
 
+func (cg *ChainGen) StateManager() *stmgr.StateManager {
+	return cg.sm
+}
+
 func (cg *ChainGen) SetStateManager(sm *stmgr.StateManager) {
 	cg.sm = sm
 }
@@ -383,15 +387,32 @@ func (cg *ChainGen) SetWinningPoStProver(m address.Address, wpp WinningPoStProve
 }
 
 func (cg *ChainGen) NextTipSetFromMiners(base *types.TipSet, miners []address.Address) (*MinedTipSet, error) {
-	var blks []*types.FullBlock
+	ms, err := cg.GetMessages(cg)
-
-	msgs, err := cg.GetMessages(cg)
 	if err != nil {
 		return nil, xerrors.Errorf("get random messages: %w", err)
 	}
 
+	msgs := make([][]*types.SignedMessage, len(miners))
+	for i := range msgs {
+		msgs[i] = ms
+	}
+
+	fts, err := cg.NextTipSetFromMinersWithMessages(base, miners, msgs)
+	if err != nil {
+		return nil, err
+	}
+
+	return &MinedTipSet{
+		TipSet:   fts,
+		Messages: ms,
+	}, nil
+}
+
+func (cg *ChainGen) NextTipSetFromMinersWithMessages(base *types.TipSet, miners []address.Address, msgs [][]*types.SignedMessage) (*store.FullTipSet, error) {
+	var blks []*types.FullBlock
+
 	for round := base.Height() + 1; len(blks) == 0; round++ {
-		for _, m := range miners {
+		for mi, m := range miners {
 			bvals, et, ticket, err := cg.nextBlockProof(context.TODO(), base, m, round)
 			if err != nil {
 				return nil, xerrors.Errorf("next block proof: %w", err)
@@ -404,7 +425,7 @@ func (cg *ChainGen) NextTipSetFromMiners(base *types.TipSet, miners []address.Ad
 					return nil, err
 				}
 
-				fblk, err := cg.makeBlock(base, m, ticket, et, bvals, round, wpost, msgs)
+				fblk, err := cg.makeBlock(base, m, ticket, et, bvals, round, wpost, msgs[mi])
 				if err != nil {
 					return nil, xerrors.Errorf("making a block for next tipset failed: %w", err)
 				}
@@ -418,12 +439,7 @@ func (cg *ChainGen) NextTipSetFromMiners(base *types.TipSet, miners []address.Ad
 		}
 	}
 
-	fts := store.NewFullTipSet(blks)
+	return store.NewFullTipSet(blks), nil
-
-	return &MinedTipSet{
-		TipSet:   fts,
-		Messages: msgs,
-	}, nil
 }
 
 func (cg *ChainGen) makeBlock(parents *types.TipSet, m address.Address, vrfticket *types.Ticket,

chain/stmgr/call.go

@@ -196,7 +196,7 @@ func (sm *StateManager) Replay(ctx context.Context, ts *types.TipSet, mcid cid.C
 	var outm *types.Message
 	var outr *vm.ApplyRet
 
-	_, _, err := sm.computeTipSetState(ctx, ts.Blocks(), func(c cid.Cid, m *types.Message, ret *vm.ApplyRet) error {
+	_, _, err := sm.computeTipSetState(ctx, ts, func(c cid.Cid, m *types.Message, ret *vm.ApplyRet) error {
 		if c == mcid {
 			outm = m
 			outr = ret

chain/stmgr/stmgr.go

@@ -111,7 +111,7 @@ func (sm *StateManager) TipSetState(ctx context.Context, ts *types.TipSet) (st c
 		return ts.Blocks()[0].ParentStateRoot, ts.Blocks()[0].ParentMessageReceipts, nil
 	}
 
-	st, rec, err = sm.computeTipSetState(ctx, ts.Blocks(), nil)
+	st, rec, err = sm.computeTipSetState(ctx, ts, nil)
 	if err != nil {
 		return cid.Undef, cid.Undef, err
 	}
@@ -121,7 +121,7 @@ func (sm *StateManager) TipSetState(ctx context.Context, ts *types.TipSet) (st c
 
 func (sm *StateManager) ExecutionTrace(ctx context.Context, ts *types.TipSet) (cid.Cid, []*api.InvocResult, error) {
 	var trace []*api.InvocResult
-	st, _, err := sm.computeTipSetState(ctx, ts.Blocks(), func(mcid cid.Cid, msg *types.Message, ret *vm.ApplyRet) error {
+	st, _, err := sm.computeTipSetState(ctx, ts, func(mcid cid.Cid, msg *types.Message, ret *vm.ApplyRet) error {
 		ir := &api.InvocResult{
 			Msg:            msg,
 			MsgRct:         &ret.MessageReceipt,
@@ -141,16 +141,9 @@ func (sm *StateManager) ExecutionTrace(ctx context.Context, ts *types.TipSet) (c
 	return st, trace, nil
 }
 
-type BlockMessages struct {
-	Miner         address.Address
-	BlsMessages   []types.ChainMsg
-	SecpkMessages []types.ChainMsg
-	WinCount      int64
-}
-
 type ExecCallback func(cid.Cid, *types.Message, *vm.ApplyRet) error
 
-func (sm *StateManager) ApplyBlocks(ctx context.Context, parentEpoch abi.ChainEpoch, pstate cid.Cid, bms []BlockMessages, epoch abi.ChainEpoch, r vm.Rand, cb ExecCallback, baseFee abi.TokenAmount) (cid.Cid, cid.Cid, error) {
+func (sm *StateManager) ApplyBlocks(ctx context.Context, parentEpoch abi.ChainEpoch, pstate cid.Cid, bms []store.BlockMessages, epoch abi.ChainEpoch, r vm.Rand, cb ExecCallback, baseFee abi.TokenAmount) (cid.Cid, cid.Cid, error) {
 
 	vmopt := &vm.VMOpts{
 		StateBase:  pstate,
@@ -311,10 +304,12 @@ func (sm *StateManager) ApplyBlocks(ctx context.Context, parentEpoch abi.ChainEp
 	return st, rectroot, nil
 }
 
-func (sm *StateManager) computeTipSetState(ctx context.Context, blks []*types.BlockHeader, cb ExecCallback) (cid.Cid, cid.Cid, error) {
+func (sm *StateManager) computeTipSetState(ctx context.Context, ts *types.TipSet, cb ExecCallback) (cid.Cid, cid.Cid, error) {
 	ctx, span := trace.StartSpan(ctx, "computeTipSetState")
 	defer span.End()
 
+	blks := ts.Blocks()
+
 	for i := 0; i < len(blks); i++ {
 		for j := i + 1; j < len(blks); j++ {
 			if blks[i].Miner == blks[j].Miner {
@@ -343,30 +338,11 @@ func (sm *StateManager) computeTipSetState(ctx context.Context, blks []*types.Bl
 
 	r := store.NewChainRand(sm.cs, cids, blks[0].Height)
 
-	var blkmsgs []BlockMessages
+	blkmsgs, err := sm.cs.BlockMsgsForTipset(ts)
-	for _, b := range blks {
+	if err != nil {
-		bms, sms, err := sm.cs.MessagesForBlock(b)
+		return cid.Undef, cid.Undef, xerrors.Errorf("getting block messages for tipset: %w", err)
-		if err != nil {
-			return cid.Undef, cid.Undef, xerrors.Errorf("failed to get messages for block: %w", err)
-		}
-
-		bm := BlockMessages{
-			Miner:         b.Miner,
-			BlsMessages:   make([]types.ChainMsg, 0, len(bms)),
-			SecpkMessages: make([]types.ChainMsg, 0, len(sms)),
-			WinCount:      b.ElectionProof.WinCount,
-		}
-
-		for _, m := range bms {
-			bm.BlsMessages = append(bm.BlsMessages, m)
-		}
-
-		for _, m := range sms {
-			bm.SecpkMessages = append(bm.SecpkMessages, m)
-		}
-
-		blkmsgs = append(blkmsgs, bm)
 	}
+
 	baseFee := blks[0].ParentBaseFee
 
 	return sm.ApplyBlocks(ctx, parentEpoch, pstate, blkmsgs, blks[0].Height, r, cb, baseFee)

chain/store/store.go

@@ -707,9 +707,15 @@ func (cs *ChainStore) readAMTCids(root cid.Cid) ([]cid.Cid, error) {
 	return cids, nil
 }
 
-func (cs *ChainStore) MessagesForTipset(ts *types.TipSet) ([]types.ChainMsg, error) {
+type BlockMessages struct {
+	Miner         address.Address
+	BlsMessages   []types.ChainMsg
+	SecpkMessages []types.ChainMsg
+	WinCount      int64
+}
+
+func (cs *ChainStore) BlockMsgsForTipset(ts *types.TipSet) ([]BlockMessages, error) {
 	applied := make(map[address.Address]uint64)
-	balances := make(map[address.Address]types.BigInt)
 
 	cst := cbor.NewCborStore(cs.bs)
 	st, err := state.LoadStateTree(cst, ts.Blocks()[0].ParentStateRoot)
@@ -725,43 +731,80 @@ func (cs *ChainStore) MessagesForTipset(ts *types.TipSet) ([]types.ChainMsg, err
 			}
 
 			applied[a] = act.Nonce
-			balances[a] = act.Balance
 		}
 		return nil
 	}
 
-	var out []types.ChainMsg
+	selectMsg := func(m *types.Message) (bool, error) {
+		if err := preloadAddr(m.From); err != nil {
+			return false, err
+		}
+
+		if applied[m.From] != m.Nonce {
+			return false, nil
+		}
+		applied[m.From]++
+
+		return true, nil
+	}
+
+	var out []BlockMessages
 	for _, b := range ts.Blocks() {
+
 		bms, sms, err := cs.MessagesForBlock(b)
 		if err != nil {
 			return nil, xerrors.Errorf("failed to get messages for block: %w", err)
 		}
 
-		cmsgs := make([]types.ChainMsg, 0, len(bms)+len(sms))
+		bm := BlockMessages{
-		for _, m := range bms {
+			Miner:         b.Miner,
-			cmsgs = append(cmsgs, m)
+			BlsMessages:   make([]types.ChainMsg, 0, len(bms)),
-		}
+			SecpkMessages: make([]types.ChainMsg, 0, len(sms)),
-		for _, sm := range sms {
+			WinCount:      b.ElectionProof.WinCount,
-			cmsgs = append(cmsgs, sm)
 		}
 
-		for _, cm := range cmsgs {
+		for _, bmsg := range bms {
-			m := cm.VMMessage()
+			b, err := selectMsg(bmsg.VMMessage())
-			if err := preloadAddr(m.From); err != nil {
+			if err != nil {
-				return nil, err
+				return nil, xerrors.Errorf("failed to decide whether to select message for block: %w", err)
 			}
 
-			if applied[m.From] != m.Nonce {
+			if b {
-				continue
+				bm.BlsMessages = append(bm.BlsMessages, bmsg)
 			}
-			applied[m.From]++
+		}
 
-			if balances[m.From].LessThan(m.RequiredFunds()) {
+		for _, smsg := range sms {
-				continue
+			b, err := selectMsg(smsg.VMMessage())
+			if err != nil {
+				return nil, xerrors.Errorf("failed to decide whether to select message for block: %w", err)
 			}
-			balances[m.From] = types.BigSub(balances[m.From], m.RequiredFunds())
 
-			out = append(out, cm)
+			if b {
+				bm.SecpkMessages = append(bm.SecpkMessages, smsg)
+			}
+		}
+
+		out = append(out, bm)
+	}
+
+	return out, nil
+}
+
+func (cs *ChainStore) MessagesForTipset(ts *types.TipSet) ([]types.ChainMsg, error) {
+	bmsgs, err := cs.BlockMsgsForTipset(ts)
+	if err != nil {
+		return nil, err
+	}
+
+	var out []types.ChainMsg
+	for _, bm := range bmsgs {
+		for _, blsm := range bm.BlsMessages {
+			out = append(out, blsm)
+		}
+
+		for _, secm := range bm.SecpkMessages {
+			out = append(out, secm)
 		}
 	}
 

chain/sync_test.go

@@ -3,6 +3,7 @@ package chain_test
 import (
 	"context"
 	"fmt"
+	"github.com/ipfs/go-cid"
 	"os"
 	"testing"
 	"time"
@@ -172,7 +173,7 @@ func (tu *syncTestUtil) pushTsExpectErr(to int, fts *store.FullTipSet, experr bo
 	}
 }
 
-func (tu *syncTestUtil) mineOnBlock(blk *store.FullTipSet, to int, miners []int, wait, fail bool) *store.FullTipSet {
+func (tu *syncTestUtil) mineOnBlock(blk *store.FullTipSet, to int, miners []int, wait, fail bool, msgs [][]*types.SignedMessage) *store.FullTipSet {
 	if miners == nil {
 		for i := range tu.g.Miners {
 			miners = append(miners, i)
@@ -186,20 +187,28 @@ func (tu *syncTestUtil) mineOnBlock(blk *store.FullTipSet, to int, miners []int,
 
 	fmt.Println("Miner mining block: ", maddrs)
 
-	mts, err := tu.g.NextTipSetFromMiners(blk.TipSet(), maddrs)
+	var nts *store.FullTipSet
-	require.NoError(tu.t, err)
+	var err error
-
+	if msgs != nil {
-	if fail {
+		nts, err = tu.g.NextTipSetFromMinersWithMessages(blk.TipSet(), maddrs, msgs)
-		tu.pushTsExpectErr(to, mts.TipSet, true)
+		require.NoError(tu.t, err)
 	} else {
-		tu.pushFtsAndWait(to, mts.TipSet, wait)
+		mt, err := tu.g.NextTipSetFromMiners(blk.TipSet(), maddrs)
+		require.NoError(tu.t, err)
+		nts = mt.TipSet
 	}
 
-	return mts.TipSet
+	if fail {
+		tu.pushTsExpectErr(to, nts, true)
+	} else {
+		tu.pushFtsAndWait(to, nts, wait)
+	}
+
+	return nts
 }
 
 func (tu *syncTestUtil) mineNewBlock(src int, miners []int) {
-	mts := tu.mineOnBlock(tu.g.CurTipset, src, miners, true, false)
+	mts := tu.mineOnBlock(tu.g.CurTipset, src, miners, true, false, nil)
 	tu.g.CurTipset = mts
 }
 
@@ -416,7 +425,7 @@ func TestSyncBadTimestamp(t *testing.T) {
 	fmt.Println("BASE: ", base.Cids())
 	tu.printHeads()
 
-	a1 := tu.mineOnBlock(base, 0, nil, false, true)
+	a1 := tu.mineOnBlock(base, 0, nil, false, true, nil)
 
 	tu.g.Timestamper = nil
 	require.NoError(t, tu.g.ResyncBankerNonce(a1.TipSet()))
@@ -425,7 +434,7 @@ func TestSyncBadTimestamp(t *testing.T) {
 
 	fmt.Println("After mine bad block!")
 	tu.printHeads()
-	a2 := tu.mineOnBlock(base, 0, nil, true, false)
+	a2 := tu.mineOnBlock(base, 0, nil, true, false, nil)
 
 	tu.waitUntilSync(0, client)
 
@@ -469,7 +478,7 @@ func TestSyncBadWinningPoSt(t *testing.T) {
 	tu.g.SetWinningPoStProver(tu.g.Miners[1], &badWpp{})
 
 	// now ensure that new blocks are not accepted
-	tu.mineOnBlock(base, client, nil, false, true)
+	tu.mineOnBlock(base, client, nil, false, true, nil)
 }
 
 func (tu *syncTestUtil) loadChainToNode(to int) {
@@ -514,16 +523,16 @@ func TestSyncFork(t *testing.T) {
 	fmt.Println("Mining base: ", base.TipSet().Cids(), base.TipSet().Height())
 
 	// The two nodes fork at this point into 'a' and 'b'
-	a1 := tu.mineOnBlock(base, p1, []int{0}, true, false)
+	a1 := tu.mineOnBlock(base, p1, []int{0}, true, false, nil)
-	a := tu.mineOnBlock(a1, p1, []int{0}, true, false)
+	a := tu.mineOnBlock(a1, p1, []int{0}, true, false, nil)
-	a = tu.mineOnBlock(a, p1, []int{0}, true, false)
+	a = tu.mineOnBlock(a, p1, []int{0}, true, false, nil)
 
 	require.NoError(t, tu.g.ResyncBankerNonce(a1.TipSet()))
 	// chain B will now be heaviest
-	b := tu.mineOnBlock(base, p2, []int{1}, true, false)
+	b := tu.mineOnBlock(base, p2, []int{1}, true, false, nil)
-	b = tu.mineOnBlock(b, p2, []int{1}, true, false)
+	b = tu.mineOnBlock(b, p2, []int{1}, true, false, nil)
-	b = tu.mineOnBlock(b, p2, []int{1}, true, false)
+	b = tu.mineOnBlock(b, p2, []int{1}, true, false, nil)
-	b = tu.mineOnBlock(b, p2, []int{1}, true, false)
+	b = tu.mineOnBlock(b, p2, []int{1}, true, false, nil)
 
 	fmt.Println("A: ", a.Cids(), a.TipSet().Height())
 	fmt.Println("B: ", b.Cids(), b.TipSet().Height())
@@ -538,6 +547,99 @@ func TestSyncFork(t *testing.T) {
 	phead()
 }
 
+// This test crafts a tipset with 2 blocks, A and B.
+// A and B both include _different_ messages from sender X with nonce N (where N is the correct nonce for X).
+// We can confirm that the state can be correctly computed, and that `MessagesForTipset` behaves as expected.
+func TestDuplicateNonce(t *testing.T) {
+	H := 10
+	tu := prepSyncTest(t, H)
+
+	base := tu.g.CurTipset
+
+	// Produce a message from the banker to the rcvr
+	makeMsg := func(rcvr address.Address) *types.SignedMessage {
+
+		ba, err := tu.nds[0].StateGetActor(context.TODO(), tu.g.Banker(), base.TipSet().Key())
+		require.NoError(t, err)
+		msg := types.Message{
+			To:   rcvr,
+			From: tu.g.Banker(),
+
+			Nonce: ba.Nonce,
+
+			Value: types.NewInt(1),
+
+			Method: 0,
+
+			GasLimit:   100_000_000,
+			GasFeeCap:  types.NewInt(0),
+			GasPremium: types.NewInt(0),
+		}
+
+		sig, err := tu.g.Wallet().Sign(context.TODO(), tu.g.Banker(), msg.Cid().Bytes())
+		require.NoError(t, err)
+
+		return &types.SignedMessage{
+			Message:   msg,
+			Signature: *sig,
+		}
+	}
+
+	msgs := make([][]*types.SignedMessage, 2)
+	// Each miner includes a message from the banker with the same nonce, but to different addresses
+	for k, _ := range msgs {
+		msgs[k] = []*types.SignedMessage{makeMsg(tu.g.Miners[k])}
+	}
+
+	ts1 := tu.mineOnBlock(base, 0, []int{0, 1}, true, false, msgs)
+
+	tu.waitUntilSyncTarget(0, ts1.TipSet())
+
+	// mine another tipset
+
+	ts2 := tu.mineOnBlock(ts1, 0, []int{0, 1}, true, false, make([][]*types.SignedMessage, 2))
+	tu.waitUntilSyncTarget(0, ts2.TipSet())
+
+	var includedMsg cid.Cid
+	var skippedMsg cid.Cid
+	r0, err0 := tu.nds[0].StateGetReceipt(context.TODO(), msgs[0][0].Cid(), ts2.TipSet().Key())
+	r1, err1 := tu.nds[0].StateGetReceipt(context.TODO(), msgs[1][0].Cid(), ts2.TipSet().Key())
+
+	if err0 == nil {
+		require.Error(t, err1, "at least one of the StateGetReceipt calls should fail")
+		require.True(t, r0.ExitCode.IsSuccess())
+		includedMsg = msgs[0][0].Message.Cid()
+		skippedMsg = msgs[1][0].Message.Cid()
+	} else {
+		require.NoError(t, err1, "both the StateGetReceipt calls should not fail")
+		require.True(t, r1.ExitCode.IsSuccess())
+		includedMsg = msgs[1][0].Message.Cid()
+		skippedMsg = msgs[0][0].Message.Cid()
+	}
+
+	_, rslts, err := tu.g.StateManager().ExecutionTrace(context.TODO(), ts1.TipSet())
+	require.NoError(t, err)
+	found := false
+	for _, v := range rslts {
+		if v.Msg.Cid() == skippedMsg {
+			t.Fatal("skipped message should not be in exec trace")
+		}
+
+		if v.Msg.Cid() == includedMsg {
+			found = true
+		}
+	}
+
+	if !found {
+		t.Fatal("included message should be in exec trace")
+	}
+
+	mft, err := tu.g.ChainStore().MessagesForTipset(ts1.TipSet())
+	require.NoError(t, err)
+	require.True(t, len(mft) == 1, "only expecting one message for this tipset")
+	require.Equal(t, includedMsg, mft[0].VMMessage().Cid(), "messages for tipset didn't contain expected message")
+}
+
 func BenchmarkSyncBasic(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		runSyncBenchLength(b, 100)

chain/validation/applier.go

@@ -71,9 +71,9 @@ func (a *Applier) ApplyTipSetMessages(epoch abi.ChainEpoch, blocks []vtypes.Bloc
 	cs := store.NewChainStore(a.stateWrapper.bs, a.stateWrapper.ds, a.syscalls)
 	sm := stmgr.NewStateManager(cs)
 
-	var bms []stmgr.BlockMessages
+	var bms []store.BlockMessages
 	for _, b := range blocks {
-		bm := stmgr.BlockMessages{
+		bm := store.BlockMessages{
 			Miner:    b.Miner,
 			WinCount: 1,
 		}