feat: implement vm message extraction

chain/indexer/integrated/processor/state.go

@@ -22,6 +22,7 @@ import (
 	poweractors "github.com/filecoin-project/lily/chain/actors/builtin/power"
 	rewardactors "github.com/filecoin-project/lily/chain/actors/builtin/reward"
 	verifregactors "github.com/filecoin-project/lily/chain/actors/builtin/verifreg"
+	"github.com/filecoin-project/lily/tasks/messageexecutions/vm"
 
 	"github.com/filecoin-project/lily/tasks"
 	// actor tasks
@@ -557,6 +558,8 @@ func MakeProcessors(api tasks.DataSource, indexerTasks []string) (*IndexerProces
 			out.TipsetsProcessors[t] = gasecontask.NewTask(api)
 		case tasktype.MultisigApproval:
 			out.TipsetsProcessors[t] = msapprovaltask.NewTask(api)
+		case tasktype.VmMessage:
+			out.TipsetsProcessors[t] = vm.NewTask(api)
 
 			//
 			// Blocks

chain/indexer/integrated/processor/state_internal_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/filecoin-project/lily/chain/actors/builtin/reward"
 	"github.com/filecoin-project/lily/chain/actors/builtin/verifreg"
 	"github.com/filecoin-project/lily/chain/indexer/tasktype"
+	"github.com/filecoin-project/lily/tasks/messageexecutions/vm"
 
 	"github.com/filecoin-project/lily/tasks/actorstate"
 	inittask "github.com/filecoin-project/lily/tasks/actorstate/init_"
@@ -48,7 +49,7 @@ func TestNewProcessor(t *testing.T) {
 	require.Equal(t, t.Name(), proc.name)
 	require.Len(t, proc.actorProcessors, 21)
 	require.Len(t, proc.tipsetProcessors, 5)
-	require.Len(t, proc.tipsetsProcessors, 9)
+	require.Len(t, proc.tipsetsProcessors, 10)
 	require.Len(t, proc.builtinProcessors, 1)
 
 	require.Equal(t, message.NewTask(nil), proc.tipsetsProcessors[tasktype.Message])
@@ -60,6 +61,7 @@ func TestNewProcessor(t *testing.T) {
 	require.Equal(t, internalparsedmessage.NewTask(nil), proc.tipsetsProcessors[tasktype.InternalParsedMessage])
 	require.Equal(t, gaseconomy.NewTask(nil), proc.tipsetsProcessors[tasktype.MessageGasEconomy])
 	require.Equal(t, msapprovals.NewTask(nil), proc.tipsetsProcessors[tasktype.MultisigApproval])
+	require.Equal(t, vm.NewTask(nil), proc.tipsetsProcessors[tasktype.VmMessage])
 
 	require.Equal(t, headers.NewTask(), proc.tipsetProcessors[tasktype.BlockHeader])
 	require.Equal(t, parents.NewTask(), proc.tipsetProcessors[tasktype.BlockParent])

chain/indexer/integrated/processor/state_test.go

@@ -347,6 +347,6 @@ func TestMakeProcessorsAllTasks(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, proc.ActorProcessors, 21)
 	require.Len(t, proc.TipsetProcessors, 5)
-	require.Len(t, proc.TipsetsProcessors, 9)
+	require.Len(t, proc.TipsetsProcessors, 10)
 	require.Len(t, proc.ReportProcessors, 1)
 }

chain/indexer/tasktype/table_tasks.go

@@ -24,6 +24,7 @@ const (
 	ParsedMessage                  = "parsed_message"
 	InternalMessage                = "internal_messages"
 	InternalParsedMessage          = "internal_parsed_messages"
+	VmMessage                      = "vm_messages"
 	MultisigTransaction            = "multisig_transaction"
 	ChainPower                     = "chain_power"
 	PowerActorClaim                = "power_actor_claim"
@@ -62,6 +63,7 @@ var AllTableTasks = []string{
 	ParsedMessage,
 	InternalMessage,
 	InternalParsedMessage,
+	VmMessage,
 	MultisigTransaction,
 	ChainPower,
 	PowerActorClaim,
@@ -100,6 +102,7 @@ var TableLookup = map[string]struct{}{
 	ParsedMessage:                  {},
 	InternalMessage:                {},
 	InternalParsedMessage:          {},
+	VmMessage:                      {},
 	MultisigTransaction:            {},
 	ChainPower:                     {},
 	PowerActorClaim:                {},
@@ -138,6 +141,7 @@ var TableComment = map[string]string{
 	ParsedMessage:                  ``,
 	InternalMessage:                ``,
 	InternalParsedMessage:          ``,
+	VmMessage:                      ``,
 	MultisigTransaction:            ``,
 	ChainPower:                     ``,
 	PowerActorClaim:                ``,
@@ -178,8 +182,9 @@ var TableFieldComments = map[string]map[string]string{
 		"DealID":               "Identifier for the deal.",
 		"EndEpoch":             "The epoch at which this deal with end.",
 		"Height":               "Epoch at which this deal proposal was added or changed.",
+		"IsString":             "When true Label contains a valid UTF-8 string encoded in base64. When false Label contains raw bytes encoded in base64. Related to FIP: https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0027.md",
 		"IsVerified":           "Deal is with a verified provider.",
-		"Label":                "An arbitrary client chosen label to apply to the deal.",
+		"Label":                "An arbitrary client chosen label to apply to the deal. The value is base64 encoded before persisting.",
 		"PaddedPieceSize":      "The piece size in bytes with padding.",
 		"PieceCID":             "CID of a sector piece. A Piece is an object that represents a whole or part of a File.",
 		"ProviderCollateral":   "The amount of FIL (in attoFIL) the provider has pledged as collateral. The Provider deal collateral is only slashed when a sector is terminated before the deal expires.",
@@ -197,6 +202,21 @@ var TableFieldComments = map[string]map[string]string{
 	ParsedMessage:         {},
 	InternalMessage:       {},
 	InternalParsedMessage: {},
+	VmMessage: {
+		"ActorCode": "ActorCode of To (receiver).",
+		"Cid":       "Cid of the message.",
+		"ExitCode":  "ExitCode of message execution.",
+		"From":      "From sender of message.",
+		"GasUsed":   "GasUsed by message.",
+		"Height":    "Height message was executed at.",
+		"Method":    "Method called on To (receiver).",
+		"Params":    "Params contained in message.",
+		"Returns":   "Return value of message.",
+		"Source":    "On-chain message triggering the message.",
+		"StateRoot": "StateRoot message was applied to.",
+		"To":        "To receiver of message.",
+		"Value":     "Value attoFIL contained in message.",
+	},
 	MultisigTransaction: {
 		"To": "Transaction State",
 	},

chain/indexer/tasktype/tasks.go

@@ -79,6 +79,7 @@ var TaskLookup = map[string][]string{
 	ImplicitMessageTask: {
 		InternalMessage,
 		InternalParsedMessage,
+		VmMessage,
 	},
 	ChainConsensusTask: {
 		ChainConsensus,

chain/indexer/tasktype/tasks_test.go

@@ -66,7 +66,7 @@ func TestMakeTaskNamesAlias(t *testing.T) {
 		},
 		{
 			taskAlias: tasktype.ImplicitMessageTask,
-			tasks:     []string{tasktype.InternalMessage, tasktype.InternalParsedMessage},
+			tasks:     []string{tasktype.InternalMessage, tasktype.InternalParsedMessage, tasktype.VmMessage},
 		},
 		{
 			taskAlias: tasktype.ChainConsensusTask,
@@ -99,9 +99,8 @@ func TestMakeAllTaskAliasNames(t *testing.T) {
 	require.Len(t, actual, len(storage.Models))
 }
 
-const TotalTableTasks = 35
-
 func TestMakeAllTaskNames(t *testing.T) {
+	const TotalTableTasks = 36
 	actual, err := tasktype.MakeTaskNames(tasktype.AllTableTasks)
 	require.NoError(t, err)
 	// if this test fails it means a new task name was added, update the above test

lens/util/repo.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"time"
 
 	"github.com/filecoin-project/go-bitfield"
 	"github.com/filecoin-project/go-state-types/exitcode"
@@ -282,6 +283,34 @@ func ParseParams(params []byte, method abi.MethodNum, actCode cid.Cid) (string,
 	return string(b), m.Name, err
 }
 
+func ParseReturn(ret []byte, method abi.MethodNum, actCode cid.Cid) (string, string, error) {
+	m, found := ActorRegistry.Methods[actCode][method]
+	if !found {
+		return "", "", fmt.Errorf("unknown method %d for actor %s", method, actCode)
+	}
+
+	// if the actor method doesn't expect returns don't parse them
+	if m.Ret == reflect.TypeOf(new(abi.EmptyValue)) {
+		return "", m.Name, nil
+	}
+
+	p := reflect.New(m.Ret.Elem()).Interface().(cbg.CBORUnmarshaler)
+	if err := p.UnmarshalCBOR(bytes.NewReader(ret)); err != nil {
+		actorName := builtin.ActorNameByCode(actCode)
+		return "", m.Name, fmt.Errorf("cbor decode into %s %s:(%s.%d) failed: %v", m.Name, actorName, actCode, method, err)
+	}
+
+	b, err := MarshalWithOverrides(p, map[reflect.Type]marshaller{
+		reflect.TypeOf(bitfield.BitField{}): bitfieldCountMarshaller,
+	})
+	if err != nil {
+		return "", "", fmt.Errorf("failed to parse message return method: %d, actor code: %s, return: %s: %w", method, actCode, string(ret), err)
+	}
+
+	return string(b), m.Name, err
+
+}
+
 func MethodAndParamsForMessage(m *types.Message, destCode cid.Cid) (string, string, error) {
 	// Method is optional, zero means a plain value transfer
 	if m.Method == 0 {
@@ -308,6 +337,71 @@ func MethodAndParamsForMessage(m *types.Message, destCode cid.Cid) (string, stri
 	return method, params, nil
 }
 
+type MessageParamsReturn struct {
+	MethodName string
+	Params     string
+	Return     string
+}
+
+func MethodParamsReturnForMessage(m *MessageTrace, destCode cid.Cid) (*MessageParamsReturn, error) {
+	// Method is optional, zero means a plain value transfer
+	if m.Message.Method == 0 {
+		return &MessageParamsReturn{
+			MethodName: "Send",
+			Params:     "",
+			Return:     "",
+		}, nil
+	}
+
+	if !destCode.Defined() {
+		return nil, fmt.Errorf("missing actor code")
+	}
+
+	params, _, err := ParseParams(m.Message.Params, m.Message.Method, destCode)
+	if err != nil {
+		log.Warnf("failed to parse parameters of message %s: %v", m.Message.Cid(), err)
+		return nil, fmt.Errorf("unknown method for actor type %s method %d: %w", destCode.String(), int64(m.Message.Method), err)
+	}
+	ret, method, err := ParseReturn(m.Receipt.Return, m.Message.Method, destCode)
+	if err != nil {
+		log.Warnf("failed to parse return of message %s: %v", m.Message.Cid(), err)
+		return nil, fmt.Errorf("unknown method for actor type %s method %d: %w", destCode.String(), int64(m.Message.Method), err)
+	}
+
+	return &MessageParamsReturn{
+		MethodName: method,
+		Params:     params,
+		Return:     ret,
+	}, nil
+}
+
+func walkExecutionTrace(et *types.ExecutionTrace, trace *[]*MessageTrace) {
+	for _, sub := range et.Subcalls {
+		*trace = append(*trace, &MessageTrace{
+			Message:   sub.Msg,
+			Receipt:   sub.MsgRct,
+			Error:     sub.Error,
+			Duration:  sub.Duration,
+			GasCharge: sub.GasCharges,
+		})
+		walkExecutionTrace(&sub, trace) //nolint:scopelint,gosec
+	}
+}
+
+type MessageTrace struct {
+	Message   *types.Message
+	Receipt   *types.MessageReceipt
+	Error     string
+	Duration  time.Duration
+	GasCharge []*types.GasTrace
+}
+
+func GetChildMessagesOf(m *lens.MessageExecution) []*MessageTrace {
+	var out []*MessageTrace
+	walkExecutionTrace(&m.Ret.ExecutionTrace, &out)
+	return out
+}
+
 func ActorNameAndFamilyFromCode(c cid.Cid) (name string, family string, err error) {
 	if !c.Defined() {
 		return "", "", fmt.Errorf("cannot derive actor name from undefined CID")

model/actors/market/dealproposal.go

@@ -49,8 +49,7 @@ type MarketDealProposal struct {
 	// An arbitrary client chosen label to apply to the deal. The value is base64 encoded before persisting.
 	Label string
 
-	// When true Label contains a valid UTF-8 string encoded in base64. When false Label contains raw bytes encoded in base64.
-	// Related to FIP: https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0027.md
+	// When true Label contains a valid UTF-8 string encoded in base64. When false Label contains raw bytes encoded in base64. Related to FIP: https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0027.md
 	IsString bool
 }
 

model/messages/vm.go

@@ -0,0 +1,65 @@
+package messages
+
+import (
+	"context"
+
+	"go.opencensus.io/tag"
+
+	"github.com/filecoin-project/lily/metrics"
+	"github.com/filecoin-project/lily/model"
+)
+
+type VMMessage struct {
+	tableName struct{} `pg:"vm_messages"` // nolint: structcheck
+
+	// Height message was executed at.
+	Height int64 `pg:",pk,notnull,use_zero"`
+	// StateRoot message was applied to.
+	StateRoot string `pg:",pk,notnull"`
+	// Cid of the message.
+	Cid string `pg:",pk,notnull"`
+	// On-chain message triggering the message.
+	Source string `pg:",pk,notnull"`
+
+	// From sender of message.
+	From string `pg:",notnull"`
+	// To receiver of message.
+	To string `pg:",notnull"`
+	// Value attoFIL contained in message.
+	Value string `pg:"type:numeric,notnull"`
+	// Method called on To (receiver).
+	Method uint64 `pg:",notnull,use_zero"`
+	// ActorCode of To (receiver).
+	ActorCode string `pg:",notnull"`
+	// ExitCode of message execution.
+	ExitCode int64 `pg:",notnull,use_zero"`
+	// GasUsed by message.
+	GasUsed int64 `pg:",notnull,use_zero"`
+	// Params contained in message.
+	Params string `pg:",type:jsonb"`
+	// Return value of message.
+	Returns string `pg:",type:jsonb"`
+}
+
+func (v *VMMessage) Persist(ctx context.Context, s model.StorageBatch, version model.Version) error {
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "vm_messages"))
+	stop := metrics.Timer(ctx, metrics.PersistDuration)
+	defer stop()
+
+	metrics.RecordCount(ctx, metrics.PersistModel, 1)
+	return s.PersistModel(ctx, v)
+}
+
+type VMMessageList []*VMMessage
+
+func (vl VMMessageList) Persist(ctx context.Context, s model.StorageBatch, version model.Version) error {
+	if len(vl) == 0 {
+		return nil
+	}
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "vm_messages"))
+	stop := metrics.Timer(ctx, metrics.PersistDuration)
+	defer stop()
+
+	metrics.RecordCount(ctx, metrics.PersistModel, len(vl))
+	return s.PersistModel(ctx, vl)
+}

schemas/v1/8_vm_messages.go

@@ -0,0 +1,46 @@
+package v1
+
+func init() {
+	patches.Register(
+		8,
+		`
+ CREATE TABLE {{ .SchemaName | default "public"}}.vm_messages (
+    height bigint NOT NULL,
+    state_root text NOT NULL,
+    cid text NOT NULL,
+	source text,
+    "from" text NOT NULL,
+    "to" text NOT NULL,
+    value numeric NOT NULL,
+    method bigint NOT NULL,
+    actor_code text NOT NULL,
+    exit_code bigint NOT NULL,
+    gas_used bigint NOT NULL,
+    params jsonb,
+	returns jsonb
+);
+ALTER TABLE ONLY {{ .SchemaName | default "public"}}.vm_messages ADD CONSTRAINT vm_messages_pkey PRIMARY KEY (height, state_root, cid, source);
+CREATE INDEX vm_messages_height_idx ON {{ .SchemaName | default "public"}}.vm_messages USING BTREE (height);
+CREATE INDEX vm_messages_cid_idx ON {{ .SchemaName | default "public"}}.vm_messages USING HASH (cid);
+CREATE INDEX vm_messages_source_idx ON {{ .SchemaName | default "public"}}.vm_messages USING HASH (source);
+CREATE INDEX vm_messages_from_idx ON {{ .SchemaName | default "public"}}.vm_messages USING HASH ("from");
+CREATE INDEX vm_messages_to_idx ON {{ .SchemaName | default "public"}}.vm_messages USING HASH ("to");
+CREATE INDEX vm_messages_actor_code_method_idx ON {{ .SchemaName | default "public"}}.vm_messages USING BTREE (actor_code, method);
+
+
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.height IS 'Height message was executed at.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.state_root IS 'CID of the parent state root at which this message was executed.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.cid IS 'CID of the message (note this CID does not appear on chain).';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.source IS 'CID of the on-chain message that caused this message to be sent.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages."from" IS 'Address of the actor that sent the message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages."to" IS 'Address of the actor that received the message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.value IS 'Amount of FIL (in attoFIL) transferred by this message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.method IS 'The method number invoked on the recipient actor. Only unique to the actor the method is being invoked on. A method number of 0 is a plain token transfer - no method execution';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.actor_code IS 'The CID of the actor that received the message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.exit_code IS 'The exit code that was returned as a result of executing the message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.gas_used IS 'A measure of the amount of resources (or units of gas) consumed, in order to execute a message.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.params IS 'Message parameters parsed and serialized as a JSON object.';
+COMMENT ON COLUMN {{ .SchemaName | default "public"}}.vm_messages.returns IS 'Result returned from executing a message parsed and serialized as a JSON object.';
+`,
+	)
+}

storage/sql.go

@@ -60,6 +60,7 @@ var Models = []interface{}{
 	(*messages.ParsedMessage)(nil),
 	(*messages.InternalMessage)(nil),
 	(*messages.InternalParsedMessage)(nil),
+	(*messages.VMMessage)(nil),
 
 	(*multisig.MultisigTransaction)(nil),