feat: Add new function for filtering invalid message

chain/datasource/datasource.go

@@ -147,6 +147,10 @@ func (t *DataSource) TipSetBlockMessages(ctx context.Context, ts *types.TipSet)
 	return t.node.MessagesForTipSetBlocks(ctx, ts)
 }
 
+func (t *DataSource) MessagesWithDeduplicationForTipSet(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error) {
+	return t.node.MessagesWithDeduplicationForTipSet(ctx, ts)
+}
+
 func (t *DataSource) StateListActors(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error) {
 	return t.node.StateListActors(ctx, tsk)
 }

lens/interface.go

@@ -51,6 +51,7 @@ type ChainAPI interface {
 
 	MessagesForTipSetBlocks(ctx context.Context, ts *types.TipSet) ([]*BlockMessages, error)
 	TipSetMessageReceipts(ctx context.Context, ts, pts *types.TipSet) ([]*BlockMessageReceipts, error)
+	MessagesWithDeduplicationForTipSet(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error)
 
 	// added during hyperspace
 	ChainGetEvents(ctx context.Context, root cid.Cid) ([]types.Event, error)

lens/lily/impl.go

@@ -592,6 +592,21 @@ func (m *LilyNodeAPI) MessagesForTipSetBlocks(ctx context.Context, ts *types.Tip
 	return out, nil
 }
 
+func (m *LilyNodeAPI) MessagesWithDeduplicationForTipSet(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error) {
+	blkMsgs, err := m.ChainAPI.Chain.BlockMsgsForTipset(ctx, ts)
+	if err != nil {
+		return nil, err
+	}
+
+	var out []types.ChainMsg
+	for _, blk := range blkMsgs {
+		out = append(out, blk.BlsMessages...)
+		out = append(out, blk.SecpkMessages...)
+	}
+
+	return out, nil
+}
+
 // TipSetMessageReceipts returns the blocks and messages in `pts` and their corresponding receipts from `ts` matching block order in tipset (`pts`).
 func (m *LilyNodeAPI) TipSetMessageReceipts(ctx context.Context, ts, pts *types.TipSet) ([]*lens.BlockMessageReceipts, error) {
 	// sanity check args

tasks/api.go

@@ -70,6 +70,7 @@ type DataSource interface {
 
 	TipSetMessageReceipts(ctx context.Context, ts, pts *types.TipSet) ([]*lens.BlockMessageReceipts, error)
 	MessageReceiptEvents(ctx context.Context, root cid.Cid) ([]types.Event, error)
+	MessagesWithDeduplicationForTipSet(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error)
 
 	DiffSectors(ctx context.Context, addr address.Address, ts, pts *types.TipSet, pre, cur miner.State) (*miner.SectorChanges, error)
 	DiffPreCommits(ctx context.Context, addr address.Address, ts, pts *types.TipSet, pre, cur miner.State) (*miner.PreCommitChanges, error)

tasks/messages/gaseconomy/task.go

@@ -16,6 +16,8 @@ import (
 	messagemodel "github.com/filecoin-project/lily/model/messages"
 	visormodel "github.com/filecoin-project/lily/model/visor"
 	"github.com/filecoin-project/lily/tasks"
+
+	logging "github.com/ipfs/go-log/v2"
 )
 
 type Task struct {
@@ -28,6 +30,8 @@ func NewTask(node tasks.DataSource) *Task {
 	}
 }
 
+var log = logging.Logger("lily/tasks/gaseconomy")
+
 func (t *Task) ProcessTipSet(ctx context.Context, current *types.TipSet) (model.Persistable, *visormodel.ProcessingReport, error) {
 	ctx, span := otel.Tracer("").Start(ctx, "ProcessTipSet")
 	if span.IsRecording() {
@@ -44,6 +48,19 @@ func (t *Task) ProcessTipSet(ctx context.Context, current *types.TipSet) (model.
 		StateRoot: current.ParentState().String(),
 	}
 
+	validMsgCid := make(map[cid.Cid]bool)
+	uniqMsg, err := t.node.MessagesWithDeduplicationForTipSet(ctx, current)
+	if err != nil {
+		log.Errorf("Error at getting messages with deduplication: %v", err)
+	}
+	if uniqMsg != nil {
+		for _, msg := range uniqMsg {
+			validMsgCid[msg.Cid()] = true
+		}
+	}
+
+	log.Infof("Get the count of valid messages: %v", len(validMsgCid))
+
 	msgrec, err := t.node.TipSetBlockMessages(ctx, current)
 	if err != nil {
 		report.ErrorsDetected = fmt.Errorf("getting tipset messages receipts: %w", err)
@@ -65,6 +82,10 @@ func (t *Task) ProcessTipSet(ctx context.Context, current *types.TipSet) (model.
 		}
 
 		for _, msg := range mr.BlsMessages {
+			if _, exists := validMsgCid[msg.Cid()]; !exists {
+				log.Errorf("Get invalid message cid: %v", msg.Cid())
+				continue
+			}
 			// calculate total gas limit of executed messages regardless of duplicates.
 			totalGasLimit += msg.GasLimit
 			if exeMsgSeen[msg.Cid()] {
@@ -76,6 +97,10 @@ func (t *Task) ProcessTipSet(ctx context.Context, current *types.TipSet) (model.
 
 		}
 		for _, msg := range mr.SecpMessages {
+			if _, exists := validMsgCid[msg.Cid()]; !exists {
+				log.Errorf("Get invalid message cid: %v", msg.Cid())
+				continue
+			}
 			// calculate total gas limit of executed messages regardless of duplicates.
 			totalGasLimit += msg.VMMessage().GasLimit
 			if exeMsgSeen[msg.Cid()] {

testutil/lens.go

@@ -51,6 +51,11 @@ func (aw *APIWrapper) TipSetMessageReceipts(ctx context.Context, ts, pts *types.
 	panic("implement me")
 }
 
+func (aw *APIWrapper) MessagesWithDeduplicationForTipSet(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (aw *APIWrapper) CirculatingSupply(ctx context.Context, key types.TipSetKey) (api.CirculatingSupply, error) {
 	return aw.StateVMCirculatingSupplyInternal(ctx, key)
 }