chain: Test sync with messages

chain/gen/gen.go

@@ -3,6 +3,7 @@ package gen
 import (
 	"bytes"
 	"context"
+	"sync/atomic"
 
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-car"
@@ -23,6 +24,8 @@ import (
 
 var log = logging.Logger("gen")
 
+const msgsPerBlock = 2
+
 type ChainGen struct {
 	accounts []address.Address
 
@@ -32,11 +35,15 @@ type ChainGen struct {
 
 	cs *store.ChainStore
 
-	genesis *types.BlockHeader
-
+	genesis  *types.BlockHeader
 	curBlock *types.FullBlock
 
-	miner address.Address
+	w *wallet.Wallet
+
+	miner       address.Address
+	receivers   []address.Address
+	banker      address.Address
+	bankerNonce uint64
 
 	r  repo.Repo
 	lr repo.LockedRepo
@@ -95,7 +102,16 @@ func NewGenerator() (*ChainGen, error) {
 		return nil, err
 	}
 
+	receievers := make([]address.Address, msgsPerBlock)
+	for r := range receievers {
+		receievers[r], err = w.GenerateKey(types.KTBLS)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	genb, err := MakeGenesisBlock(bs, map[address.Address]types.BigInt{
+		miner:  types.NewInt(5),
 		banker: types.NewInt(90000000),
 	})
 	if err != nil {
@@ -104,8 +120,6 @@ func NewGenerator() (*ChainGen, error) {
 
 	cs := store.NewChainStore(bs, ds)
 
-	msgsPerBlock := 10
-
 	genfb := &types.FullBlock{Header: genb.Genesis}
 
 	if err := cs.SetGenesis(genb.Genesis); err != nil {
@@ -117,8 +131,13 @@ func NewGenerator() (*ChainGen, error) {
 		cs:           cs,
 		msgsPerBlock: msgsPerBlock,
 		genesis:      genb.Genesis,
-		miner:        miner,
-		curBlock:     genfb,
+		w:            w,
+
+		miner:     miner,
+		banker:    banker,
+		receivers: receievers,
+
+		curBlock: genfb,
 
 		r:  mr,
 		lr: lr,
@@ -155,7 +174,45 @@ func (cg *ChainGen) NextBlock() (*types.FullBlock, error) {
 		return nil, err
 	}
 
-	var msgs []*types.SignedMessage
+	// make some transfers from banker
+
+	msgs := make([]*types.SignedMessage, cg.msgsPerBlock)
+	for m := range msgs {
+		msg := types.Message{
+			To:   cg.receivers[m],
+			From: cg.banker,
+
+			Nonce: atomic.AddUint64(&cg.bankerNonce, 1) - 1,
+
+			Value: types.NewInt(uint64(m + 1)),
+
+			Method: 0,
+
+			GasLimit: types.NewInt(10000),
+			GasPrice: types.NewInt(0),
+		}
+
+		unsigned, err := msg.Serialize()
+		if err != nil {
+			return nil, err
+		}
+
+		sig, err := cg.w.Sign(cg.banker, unsigned)
+		if err != nil {
+			return &types.FullBlock{}, err
+		}
+
+		msgs[m] = &types.SignedMessage{
+			Message:   msg,
+			Signature: *sig,
+		}
+
+		if _, err := cg.cs.PutMessage(msgs[m]); err != nil {
+			return nil, err
+		}
+	}
+
+	// create block
 
 	parents, err := types.NewTipSet([]*types.BlockHeader{cg.curBlock.Header})
 	if err != nil {

chain/sync.go

@@ -143,16 +143,6 @@ func (syncer *Syncer) InformNewHead(from peer.ID, fts *store.FullTipSet) {
 	}()
 }
 
-func (syncer *Syncer) GetPeers() []peer.ID {
-	syncer.peerHeadsLk.Lock()
-	defer syncer.peerHeadsLk.Unlock()
-	var out []peer.ID
-	for p, _ := range syncer.peerHeads {
-		out = append(out, p)
-	}
-	return out
-}
-
 func (syncer *Syncer) InformNewBlock(from peer.ID, blk *types.FullBlock) {
 	// TODO: search for other blocks that could form a tipset with this block
 	// and then send that tipset to InformNewHead

chain/sync_test.go

@@ -37,59 +37,100 @@ func repoWithChain(t *testing.T, h int) (repo.Repo, []byte) {
 	return r, genb
 }
 
-func TestSyncSimple(t *testing.T) {
-	ctx := context.Background()
+type syncTestUtil struct {
+	t *testing.T
 
-	var source api.FullNode
-	var client api.FullNode
+	ctx context.Context
+	mn  mocknet.Mocknet
+
+	genesis []byte
+
+	nds []api.FullNode
+}
+
+func (tu *syncTestUtil) addSourceNode(gen int) int {
+	if tu.genesis != nil {
+		tu.t.Fatal("source node already exists")
+	}
+
+	sourceRepo, genesis := repoWithChain(tu.t, gen)
+	var out api.FullNode
+
+	err := node.New(tu.ctx,
+		node.FullAPI(&out),
+		node.Online(),
+		node.Repo(sourceRepo),
+		node.MockHost(tu.mn),
+
+		node.Override(new(modules.Genesis), modules.LoadGenesis(genesis)),
+	)
+	require.NoError(tu.t, err)
+
+	tu.genesis = genesis
+	tu.nds = append(tu.nds, out)
+	return len(tu.nds) - 1
+}
+
+func (tu *syncTestUtil) addClientNode() int {
+	var out api.FullNode
+
+	err := node.New(tu.ctx,
+		node.FullAPI(&out),
+		node.Online(),
+		node.Repo(repo.NewMemory(nil)),
+		node.MockHost(tu.mn),
+
+		node.Override(new(modules.Genesis), modules.LoadGenesis(tu.genesis)),
+	)
+	require.NoError(tu.t, err)
+
+	tu.nds = append(tu.nds, out)
+	return len(tu.nds) - 1
+}
+
+func TestSyncSimple(t *testing.T) {
+	H := 3
+
+	ctx := context.Background()
 
 	mn := mocknet.New(ctx)
 
-	sourceRepo, genesis := repoWithChain(t, 20)
+	tu := &syncTestUtil{
+		t:   t,
+		ctx: ctx,
+		mn:  mn,
+	}
 
-	err := node.New(ctx,
-		node.FullAPI(&source),
-		node.Online(),
-		node.Repo(sourceRepo),
-		node.MockHost(mn),
+	source := tu.addSourceNode(H)
 
-		node.Override(new(modules.Genesis), modules.LoadGenesis(genesis)),
-	)
+	b, err := tu.nds[source].ChainHead(ctx)
 	require.NoError(t, err)
 
-	b, err := source.ChainHead(ctx)
-	require.NoError(t, err)
-
-	require.Equal(t, uint64(20), b.Height())
+	require.Equal(t, uint64(H), b.Height())
 	fmt.Printf("source H: %d\n", b.Height())
 
-	err = node.New(ctx,
-		node.FullAPI(&client),
-		node.Online(),
-		node.Repo(repo.NewMemory(nil)),
-		node.MockHost(mn),
+	// separate logs
+	fmt.Println("///////////////////////////////////////////////////")
 
-		node.Override(new(modules.Genesis), modules.LoadGenesis(genesis)),
-	)
-	require.NoError(t, err)
+	client := tu.addClientNode()
 
 	require.NoError(t, mn.LinkAll())
 
-	cb, err := client.ChainHead(ctx)
+	cb, err := tu.nds[client].ChainHead(ctx)
 	require.NoError(t, err)
 	require.Equal(t, uint64(0), cb.Height())
 	fmt.Printf("client H: %d\n", cb.Height())
 
-	sourcePI, err := source.NetAddrsListen(ctx)
+	sourcePI, err := tu.nds[source].NetAddrsListen(ctx)
 	require.NoError(t, err)
 
-	err = client.NetConnect(ctx, sourcePI)
+	err = tu.nds[client].NetConnect(ctx, sourcePI)
 	require.NoError(t, err)
 
-	time.Sleep(time.Second)
+	time.Sleep(time.Second * 1)
 
-	cb, err = client.ChainHead(ctx)
+	cb, err = tu.nds[client].ChainHead(ctx)
 	require.NoError(t, err)
-	require.Equal(t, uint64(20), cb.Height())
+	require.Equal(t, uint64(H), cb.Height())
 	fmt.Printf("client H: %d\n", cb.Height())
 }

chain/vm/vm.go

@@ -220,7 +220,7 @@ func (vm *VM) ApplyMessage(ctx context.Context, msg *types.Message) (*types.Mess
 	}
 
 	if msg.Nonce != fromActor.Nonce {
-		return nil, xerrors.Errorf("invalid nonce")
+		return nil, xerrors.Errorf("invalid nonce (got %d, expected %d)", msg.Nonce, fromActor.Nonce)
 	}
 	fromActor.Nonce++