Merge pull request #6642 from hunjixin/feat/add_getmessagesints_api

Add ChainGetMessagesInTipset API

.circleci/config.yml

@@ -806,6 +806,11 @@ workflows:
           suite: itest-gateway
           target: "./itests/gateway_test.go"
       
+      - test:
+          name: test-itest-get_messages_in_ts
+          suite: itest-get_messages_in_ts
+          target: "./itests/get_messages_in_ts_test.go"
+      
       - test:
           name: test-itest-multisig
           suite: itest-multisig

api/api_full.go

@@ -104,6 +104,9 @@ type FullNode interface {
 	// specified block.
 	ChainGetParentMessages(ctx context.Context, blockCid cid.Cid) ([]Message, error) //perm:read
 
+	// ChainGetMessagesInTipset returns message stores in current tipset
+	ChainGetMessagesInTipset(ctx context.Context, tsk types.TipSetKey) ([]Message, error) //perm:read
+
 	// ChainGetTipSetByHeight looks back for a tipset at the specified epoch.
 	// If there are no blocks at the specified epoch, a tipset at an earlier epoch
 	// will be returned.

api/mocks/mock_full.go

@@ -194,6 +194,21 @@ func (mr *MockFullNodeMockRecorder) ChainGetMessage(arg0, arg1 interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainGetMessage", reflect.TypeOf((*MockFullNode)(nil).ChainGetMessage), arg0, arg1)
 }
 
+// ChainGetMessagesInTipset mocks base method.
+func (m *MockFullNode) ChainGetMessagesInTipset(arg0 context.Context, arg1 types.TipSetKey) ([]api.Message, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ChainGetMessagesInTipset", arg0, arg1)
+	ret0, _ := ret[0].([]api.Message)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ChainGetMessagesInTipset indicates an expected call of ChainGetMessagesInTipset.
+func (mr *MockFullNodeMockRecorder) ChainGetMessagesInTipset(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainGetMessagesInTipset", reflect.TypeOf((*MockFullNode)(nil).ChainGetMessagesInTipset), arg0, arg1)
+}
+
 // ChainGetNode mocks base method.
 func (m *MockFullNode) ChainGetNode(arg0 context.Context, arg1 string) (*api.IpldObject, error) {
 	m.ctrl.T.Helper()

api/proxy_gen.go

@@ -127,6 +127,8 @@ type FullNodeStruct struct {
 
 		ChainGetMessage func(p0 context.Context, p1 cid.Cid) (*types.Message, error) `perm:"read"`
 
+		ChainGetMessagesInTipset func(p0 context.Context, p1 types.TipSetKey) ([]Message, error) `perm:"read"`
+
 		ChainGetNode func(p0 context.Context, p1 string) (*IpldObject, error) `perm:"read"`
 
 		ChainGetParentMessages func(p0 context.Context, p1 cid.Cid) ([]Message, error) `perm:"read"`
@@ -1101,6 +1103,14 @@ func (s *FullNodeStub) ChainGetMessage(p0 context.Context, p1 cid.Cid) (*types.M
 	return nil, xerrors.New("method not supported")
 }
 
+func (s *FullNodeStruct) ChainGetMessagesInTipset(p0 context.Context, p1 types.TipSetKey) ([]Message, error) {
+	return s.Internal.ChainGetMessagesInTipset(p0, p1)
+}
+
+func (s *FullNodeStub) ChainGetMessagesInTipset(p0 context.Context, p1 types.TipSetKey) ([]Message, error) {
+	return *new([]Message), xerrors.New("method not supported")
+}
+
 func (s *FullNodeStruct) ChainGetNode(p0 context.Context, p1 string) (*IpldObject, error) {
 	return s.Internal.ChainGetNode(p0, p1)
 }

api/v0api/full.go

@@ -92,6 +92,9 @@ type FullNode interface {
 	// specified block.
 	ChainGetParentMessages(ctx context.Context, blockCid cid.Cid) ([]api.Message, error) //perm:read
 
+	// ChainGetMessagesInTipset returns message stores in current tipset
+	ChainGetMessagesInTipset(ctx context.Context, tsk types.TipSetKey) ([]api.Message, error) //perm:read
+
 	// ChainGetTipSetByHeight looks back for a tipset at the specified epoch.
 	// If there are no blocks at the specified epoch, a tipset at an earlier epoch
 	// will be returned.

api/v0api/proxy_gen.go

@@ -45,6 +45,8 @@ type FullNodeStruct struct {
 
 		ChainGetMessage func(p0 context.Context, p1 cid.Cid) (*types.Message, error) `perm:"read"`
 
+		ChainGetMessagesInTipset func(p0 context.Context, p1 types.TipSetKey) ([]api.Message, error) `perm:"read"`
+
 		ChainGetNode func(p0 context.Context, p1 string) (*api.IpldObject, error) `perm:"read"`
 
 		ChainGetParentMessages func(p0 context.Context, p1 cid.Cid) ([]api.Message, error) `perm:"read"`
@@ -514,6 +516,14 @@ func (s *FullNodeStub) ChainGetMessage(p0 context.Context, p1 cid.Cid) (*types.M
 	return nil, xerrors.New("method not supported")
 }
 
+func (s *FullNodeStruct) ChainGetMessagesInTipset(p0 context.Context, p1 types.TipSetKey) ([]api.Message, error) {
+	return s.Internal.ChainGetMessagesInTipset(p0, p1)
+}
+
+func (s *FullNodeStub) ChainGetMessagesInTipset(p0 context.Context, p1 types.TipSetKey) ([]api.Message, error) {
+	return *new([]api.Message), xerrors.New("method not supported")
+}
+
 func (s *FullNodeStruct) ChainGetNode(p0 context.Context, p1 string) (*api.IpldObject, error) {
 	return s.Internal.ChainGetNode(p0, p1)
 }

api/v0api/v0mocks/mock_full.go

@@ -194,6 +194,21 @@ func (mr *MockFullNodeMockRecorder) ChainGetMessage(arg0, arg1 interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainGetMessage", reflect.TypeOf((*MockFullNode)(nil).ChainGetMessage), arg0, arg1)
 }
 
+// ChainGetMessagesInTipset mocks base method.
+func (m *MockFullNode) ChainGetMessagesInTipset(arg0 context.Context, arg1 types.TipSetKey) ([]api.Message, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ChainGetMessagesInTipset", arg0, arg1)
+	ret0, _ := ret[0].([]api.Message)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ChainGetMessagesInTipset indicates an expected call of ChainGetMessagesInTipset.
+func (mr *MockFullNodeMockRecorder) ChainGetMessagesInTipset(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChainGetMessagesInTipset", reflect.TypeOf((*MockFullNode)(nil).ChainGetMessagesInTipset), arg0, arg1)
+}
+
 // ChainGetNode mocks base method.
 func (m *MockFullNode) ChainGetNode(arg0 context.Context, arg1 string) (*api.IpldObject, error) {
 	m.ctrl.T.Helper()

documentation/en/api-v0-methods.md

@@ -17,6 +17,7 @@
   * [ChainGetBlockMessages](#ChainGetBlockMessages)
   * [ChainGetGenesis](#ChainGetGenesis)
   * [ChainGetMessage](#ChainGetMessage)
+  * [ChainGetMessagesInTipset](#ChainGetMessagesInTipset)
   * [ChainGetNode](#ChainGetNode)
   * [ChainGetParentMessages](#ChainGetParentMessages)
   * [ChainGetParentReceipts](#ChainGetParentReceipts)
@@ -533,6 +534,28 @@ Response:
 }
 ```
 
+### ChainGetMessagesInTipset
+ChainGetMessagesInTipset returns message stores in current tipset
+
+
+Perms: read
+
+Inputs:
+```json
+[
+  [
+    {
+      "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+    },
+    {
+      "/": "bafy2bzacebp3shtrn43k7g3unredz7fxn4gj533d3o43tqn2p2ipxxhrvchve"
+    }
+  ]
+]
+```
+
+Response: `null`
+
 ### ChainGetNode
 
 

documentation/en/api-v1-unstable-methods.md

@@ -17,6 +17,7 @@
   * [ChainGetBlockMessages](#ChainGetBlockMessages)
   * [ChainGetGenesis](#ChainGetGenesis)
   * [ChainGetMessage](#ChainGetMessage)
+  * [ChainGetMessagesInTipset](#ChainGetMessagesInTipset)
   * [ChainGetNode](#ChainGetNode)
   * [ChainGetParentMessages](#ChainGetParentMessages)
   * [ChainGetParentReceipts](#ChainGetParentReceipts)
@@ -535,6 +536,28 @@ Response:
 }
 ```
 
+### ChainGetMessagesInTipset
+ChainGetMessagesInTipset returns message stores in current tipset
+
+
+Perms: read
+
+Inputs:
+```json
+[
+  [
+    {
+      "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+    },
+    {
+      "/": "bafy2bzacebp3shtrn43k7g3unredz7fxn4gj533d3o43tqn2p2ipxxhrvchve"
+    }
+  ]
+]
+```
+
+Response: `null`
+
 ### ChainGetNode
 
 

itests/get_messages_in_ts_test.go

@@ -0,0 +1,104 @@
+package itests
+
+import (
+	"context"
+	"testing"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/itests/kit"
+	"github.com/stretchr/testify/require"
+
+	"time"
+
+	"github.com/filecoin-project/go-state-types/big"
+)
+
+func TestChainGetMessagesInTs(t *testing.T) {
+	ctx := context.Background()
+
+	kit.QuietMiningLogs()
+
+	client, _, ens := kit.EnsembleMinimal(t, kit.MockProofs())
+	ens.InterconnectAll().BeginMining(10 * time.Millisecond)
+
+	// create a new address where to send funds.
+	addr, err := client.WalletNew(ctx, types.KTBLS)
+	require.NoError(t, err)
+
+	// get the existing balance from the default wallet to then split it.
+	bal, err := client.WalletBalance(ctx, client.DefaultKey.Address)
+	require.NoError(t, err)
+
+	const iterations = 100
+
+	// we'll send half our balance (saving the other half for gas),
+	// in `iterations` increments.
+	toSend := big.Div(bal, big.NewInt(2))
+	each := big.Div(toSend, big.NewInt(iterations))
+
+	waitAllCh := make(chan struct{})
+	go func() {
+		headChangeCh, err := client.ChainNotify(ctx)
+		require.NoError(t, err)
+		<-headChangeCh //skip hccurrent
+
+		count := 0
+		for {
+			select {
+			case headChanges := <-headChangeCh:
+				for _, change := range headChanges {
+					if change.Type == store.HCApply {
+						msgs, err := client.ChainGetMessagesInTipset(ctx, change.Val.Key())
+						require.NoError(t, err)
+						count += len(msgs)
+						if count == iterations {
+							waitAllCh <- struct{}{}
+						}
+					}
+				}
+			}
+		}
+	}()
+
+	var sms []*types.SignedMessage
+	for i := 0; i < iterations; i++ {
+		msg := &types.Message{
+			From:  client.DefaultKey.Address,
+			To:    addr,
+			Value: each,
+		}
+
+		sm, err := client.MpoolPushMessage(ctx, msg, nil)
+		require.NoError(t, err)
+		require.EqualValues(t, i, sm.Message.Nonce)
+
+		sms = append(sms, sm)
+	}
+
+	select {
+	case <-waitAllCh:
+	case <-time.After(time.Minute):
+		t.Errorf("timeout to wait for pack messages")
+	}
+
+	for _, sm := range sms {
+		msgLookup, err := client.StateWaitMsg(ctx, sm.Cid(), 3, api.LookbackNoLimit, true)
+		require.NoError(t, err)
+
+		ts, err := client.ChainGetTipSet(ctx, msgLookup.TipSet)
+		require.NoError(t, err)
+
+		msgs, err := client.ChainGetMessagesInTipset(ctx, ts.Parents())
+		require.NoError(t, err)
+
+		var found bool
+		for _, msg := range msgs {
+			if msg.Cid == sm.Cid() {
+				found = true
+			}
+		}
+		require.EqualValues(t, true, found, "expect got message in tipset %v", msgLookup.TipSet)
+	}
+}

node/impl/full/chain.go

@@ -228,6 +228,33 @@ func (a *ChainAPI) ChainGetParentReceipts(ctx context.Context, bcid cid.Cid) ([]
 	return out, nil
 }
 
+func (a *ChainAPI) ChainGetMessagesInTipset(ctx context.Context, tsk types.TipSetKey) ([]api.Message, error) {
+	ts, err := a.Chain.GetTipSetFromKey(tsk)
+	if err != nil {
+		return nil, err
+	}
+
+	// genesis block has no parent messages...
+	if ts.Height() == 0 {
+		return nil, nil
+	}
+
+	cm, err := a.Chain.MessagesForTipset(ts)
+	if err != nil {
+		return nil, err
+	}
+
+	var out []api.Message
+	for _, m := range cm {
+		out = append(out, api.Message{
+			Cid:     m.Cid(),
+			Message: m.VMMessage(),
+		})
+	}
+
+	return out, nil
+}
+
 func (m *ChainModule) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
 	ts, err := m.Chain.GetTipSetFromKey(tsk)
 	if err != nil {