feat: add builtin-actor-event task

chain/datasource/datasource.go

@@ -172,6 +172,10 @@ func (t *DataSource) EthGetTransactionReceipt(ctx context.Context, txHash ethtyp
 	return t.node.EthGetTransactionReceipt(ctx, txHash)
 }
 
+func (t *DataSource) GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+	return t.node.GetActorEventsRaw(ctx, filter)
+}
+
 // TipSetMessageReceipts returns the blocks and messages in `pts` and their corresponding receipts from `ts` matching block order in tipset (`pts`).
 // TODO replace with lotus chainstore method when /~https://github.com/filecoin-project/lotus/pull/9186 lands
 func (t *DataSource) TipSetMessageReceipts(ctx context.Context, ts, pts *types.TipSet) ([]*lens.BlockMessageReceipts, error) {

chain/indexer/integrated/processor/state.go

@@ -31,6 +31,7 @@ import (
 	"github.com/filecoin-project/lily/tasks"
 	"github.com/filecoin-project/lily/tasks/messageexecutions/vm"
 	"github.com/filecoin-project/lily/tasks/messages/actorevent"
+	"github.com/filecoin-project/lily/tasks/messages/builtinactorevent"
 	"github.com/filecoin-project/lily/tasks/messages/messageparam"
 	"github.com/filecoin-project/lily/tasks/messages/receiptreturn"
 
@@ -779,6 +780,8 @@ func MakeProcessors(api tasks.DataSource, indexerTasks []string) (*IndexerProces
 			out.TipsetsProcessors[t] = vm.NewTask(api)
 		case tasktype.ActorEvent:
 			out.TipsetsProcessors[t] = actorevent.NewTask(api)
+		case tasktype.BuiltInActorEvent:
+			out.TipsetsProcessors[t] = builtinactorevent.NewTask(api)
 		case tasktype.ReceiptReturn:
 			out.TipsetsProcessors[t] = receiptreturn.NewTask(api)
 

chain/indexer/integrated/processor/state_internal_test.go

@@ -52,7 +52,7 @@ func TestNewProcessor(t *testing.T) {
 	require.Equal(t, t.Name(), proc.name)
 	require.Len(t, proc.actorProcessors, 25)
 	require.Len(t, proc.tipsetProcessors, 10)
-	require.Len(t, proc.tipsetsProcessors, 14)
+	require.Len(t, proc.tipsetsProcessors, 15)
 	require.Len(t, proc.builtinProcessors, 1)
 
 	require.Equal(t, gasoutput.NewTask(nil), proc.tipsetsProcessors[tasktype.GasOutputs])

chain/indexer/integrated/processor/state_test.go

@@ -441,6 +441,6 @@ func TestMakeProcessorsAllTasks(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, proc.ActorProcessors, 25)
 	require.Len(t, proc.TipsetProcessors, 10)
-	require.Len(t, proc.TipsetsProcessors, 14)
+	require.Len(t, proc.TipsetsProcessors, 15)
 	require.Len(t, proc.ReportProcessors, 1)
 }

chain/indexer/tasktype/table_tasks.go

@@ -53,6 +53,7 @@ const (
 	FEVMTrace                      = "fevm_traces"
 	FEVMActorDump                  = "fevm_actor_dumps"
 	MinerActorDump                 = "miner_actor_dumps"
+	BuiltInActorEvent              = "builtin_actor_event"
 )
 
 var AllTableTasks = []string{
@@ -107,6 +108,7 @@ var AllTableTasks = []string{
 	FEVMTrace,
 	FEVMActorDump,
 	MinerActorDump,
+	BuiltInActorEvent,
 }
 
 var TableLookup = map[string]struct{}{
@@ -161,6 +163,7 @@ var TableLookup = map[string]struct{}{
 	FEVMTrace:                      {},
 	FEVMActorDump:                  {},
 	MinerActorDump:                 {},
+	BuiltInActorEvent:              {},
 }
 
 var TableComment = map[string]string{
@@ -215,6 +218,7 @@ var TableComment = map[string]string{
 	FEVMTrace:                      ``,
 	FEVMActorDump:                  ``,
 	MinerActorDump:                 ``,
+	BuiltInActorEvent:              ``,
 }
 
 var TableFieldComments = map[string]map[string]string{
@@ -442,4 +446,5 @@ var TableFieldComments = map[string]map[string]string{
 		"RawBytePower":     "Claims",
 		"TotalLockedFunds": "Locked Funds",
 	},
+	BuiltInActorEvent: {},
 }

chain/indexer/tasktype/tasks.go

@@ -78,6 +78,7 @@ var TaskLookup = map[string][]string{
 		ActorEvent,
 		MessageParam,
 		ReceiptReturn,
+		BuiltInActorEvent,
 	},
 	ChainEconomicsTask: {
 		ChainEconomics,

chain/indexer/tasktype/tasks_test.go

@@ -55,7 +55,8 @@ func TestMakeTaskNamesAlias(t *testing.T) {
 		},
 		{
 			taskAlias: tasktype.MessagesTask,
-			tasks:     []string{tasktype.Message, tasktype.ParsedMessage, tasktype.Receipt, tasktype.GasOutputs, tasktype.MessageGasEconomy, tasktype.BlockMessage, tasktype.ActorEvent, tasktype.MessageParam, tasktype.ReceiptReturn},
+			tasks: []string{tasktype.Message, tasktype.ParsedMessage, tasktype.Receipt, tasktype.GasOutputs, tasktype.MessageGasEconomy, tasktype.BlockMessage, tasktype.ActorEvent, tasktype.MessageParam, tasktype.ReceiptReturn,
+				tasktype.BuiltInActorEvent},
 		},
 		{
 			taskAlias: tasktype.ChainEconomicsTask,
@@ -101,7 +102,7 @@ func TestMakeAllTaskAliasNames(t *testing.T) {
 }
 
 func TestMakeAllTaskNames(t *testing.T) {
-	const TotalTableTasks = 51
+	const TotalTableTasks = 52
 	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

go.mod

@@ -65,6 +65,7 @@ require (
 	github.com/DataDog/zstd v1.4.5
 	github.com/cenkalti/backoff/v4 v4.2.1
 	github.com/filecoin-project/go-amt-ipld/v4 v4.2.0
+	github.com/fxamacker/cbor/v2 v2.6.0
 	github.com/hibiken/asynq v0.23.0
 	github.com/hibiken/asynq/x v0.0.0-20220413130846-5c723f597e01
 	github.com/ipfs/go-ipld-format v0.6.0
@@ -315,6 +316,7 @@ require (
 	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
 	github.com/whyrusleeping/ledger-filecoin-go v0.9.1-0.20201010031517-c3dcc1bddce4 // indirect
 	github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 // indirect
+	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
 	github.com/zondax/hid v0.9.1 // indirect
 	github.com/zondax/ledger-go v0.12.1 // indirect

go.sum

@@ -391,6 +391,8 @@ github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
+github.com/fxamacker/cbor/v2 v2.6.0 h1:sU6J2usfADwWlYDAFhZBQ6TnLFBHxgesMrQfQgk1tWA=
+github.com/fxamacker/cbor/v2 v2.6.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ=
 github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
 github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
 github.com/gammazero/workerpool v1.1.3 h1:WixN4xzukFoN0XSeXF6puqEqFTl2mECI9S6W44HWy9Q=
@@ -1702,6 +1704,8 @@ github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9/go.mod h1:j4l84
 github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 h1:E9S12nwJwEOXe2d6gT6qxdvqMnNq+VnSsKPgm2ZZNds=
 github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7/go.mod h1:X2c0RVCI1eSUFI8eLcY3c0423ykwiUdxLJtkDvruhjI=
 github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
+github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
+github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xorcare/golden v0.6.0/go.mod h1:7T39/ZMvaSEZlBPoYfVFmsBLmUl3uz9IuzWj/U6FtvQ=
 github.com/xorcare/golden v0.6.1-0.20191112154924-b87f686d7542 h1:oWgZJmC1DorFZDpfMfWg7xk29yEOZiXmo/wZl+utTI8=

lens/interface.go

@@ -24,6 +24,7 @@ type API interface {
 	StateAPI
 	VMAPI
 	EthModuleAPI
+	ActorEventAPI
 
 	GetMessageExecutionsForTipSet(ctx context.Context, ts, pts *types.TipSet) ([]*MessageExecution, error)
 }
@@ -86,6 +87,10 @@ type EthModuleAPI interface {
 	EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error)
 }
 
+type ActorEventAPI interface {
+	GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)
+}
+
 type MessageExecution struct {
 	Cid       cid.Cid
 	StateRoot cid.Cid

lens/lily/api.go

@@ -62,6 +62,7 @@ type LilyAPI interface {
 	ChainGetMessagesInTipset(ctx context.Context, tsk types.TipSetKey) ([]api.Message, error)                    //perm:read
 	EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error)              //perm:read
 	StateListActors(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error)                         //perm:read
+	GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)          //perm:read
 
 	// SyncIncomingBlocks returns a channel streaming incoming, potentially not
 	// yet synced block headers.

lens/lily/impl.go

@@ -59,6 +59,7 @@ type LilyNodeAPI struct {
 	full.StateAPI
 	full.SyncAPI
 	full.EthModuleAPI
+	full.ActorEventAPI
 	common.CommonAPI
 	Events    *events.Events
 	Scheduler *schedule.Scheduler
@@ -575,6 +576,10 @@ func (m *LilyNodeAPI) StateListActors(ctx context.Context, tsk types.TipSetKey)
 	return m.StateAPI.StateListActors(ctx, tsk)
 }
 
+func (m *LilyNodeAPI) GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+	return m.ActorEventAPI.GetActorEventsRaw(ctx, filter)
+}
+
 // MessagesForTipSetBlocks returns messages stored in the blocks of the specified tipset, messages may be duplicated
 // across the returned set of BlockMessages.
 func (m *LilyNodeAPI) MessagesForTipSetBlocks(ctx context.Context, ts *types.TipSet) ([]*lens.BlockMessages, error) {

lens/lily/struct.go

@@ -71,6 +71,7 @@ type LilyAPIStruct struct {
 		ChainGetMessagesInTipset  func(ctx context.Context, tsk types.TipSetKey) ([]api.Message, error)                           `perm:"read"`
 		EthGetTransactionByHash   func(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error)                    `perm:"read"`
 		StateListActors           func(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error)                       `perm:"read"`
+		GetActorEventsRaw         func(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)          `perm:"read"`
 
 		// SyncIncomingBlocks returns a channel streaming incoming, potentially not
 		// yet synced block headers.
@@ -304,3 +305,7 @@ func (s *LilyAPIStruct) EthGetTransactionByHash(ctx context.Context, txHash *eth
 func (s *LilyAPIStruct) SyncIncomingBlocks(ctx context.Context) (<-chan *types.BlockHeader, error) {
 	return s.Internal.SyncIncomingBlocks(ctx)
 }
+
+func (s *LilyAPIStruct) GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+	return s.Internal.GetActorEventsRaw(ctx, filter)
+}

model/actors/builtinactor/builtinactorevents.go

@@ -0,0 +1,43 @@
+package builtinactor
+
+import (
+	"context"
+
+	"go.opencensus.io/tag"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/filecoin-project/lily/metrics"
+	"github.com/filecoin-project/lily/model"
+)
+
+type BuiltInActorEvent struct {
+	tableName struct{} `pg:"builtin_actor_events"` // nolint: structcheck
+
+	Height       int64  `pg:",pk,notnull,use_zero"`
+	Cid          string `pg:",pk,notnull"`
+	Emitter      string `pg:",pk,notnull"`
+	EventType    string `pg:",pk,notnull"`
+	EventEntries string `pg:",type:jsonb"`
+	EventPayload string `pg:",type:jsonb"`
+}
+
+func (ds *BuiltInActorEvent) Persist(ctx context.Context, s model.StorageBatch, _ model.Version) error {
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "builtin_actor_events"))
+	metrics.RecordCount(ctx, metrics.PersistModel, 1)
+	return s.PersistModel(ctx, ds)
+}
+
+type BuiltInActorEvents []*BuiltInActorEvent
+
+func (dss BuiltInActorEvents) Persist(ctx context.Context, s model.StorageBatch, _ model.Version) error {
+	ctx, span := otel.Tracer("").Start(ctx, "BuiltInActorEvents.Persist")
+	if span.IsRecording() {
+		span.SetAttributes(attribute.Int("count", len(dss)))
+	}
+	defer span.End()
+
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "builtin_actor_events"))
+	metrics.RecordCount(ctx, metrics.PersistModel, len(dss))
+	return s.PersistModel(ctx, dss)
+}

schemas/v1/36_builtin_actor_event.go

@@ -0,0 +1,18 @@
+package v1
+
+func init() {
+	patches.Register(
+		36,
+		`
+	CREATE TABLE IF NOT EXISTS {{ .SchemaName | default "public"}}.builtin_actor_events  (
+	    height               BIGINT NOT NULL,
+		cid                  TEXT,
+		emitter          	 TEXT,
+		event_type           TEXT,
+		event_entries        JSONB,
+		event_payload        JSONB,
+		PRIMARY KEY(height, cid, emitter, event_type)
+	);
+`,
+	)
+}

storage/sql.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/filecoin-project/lily/model"
 	"github.com/filecoin-project/lily/model/actordumps"
+	"github.com/filecoin-project/lily/model/actors/builtinactor"
 	"github.com/filecoin-project/lily/model/actors/common"
 	"github.com/filecoin-project/lily/model/actors/datacap"
 	init_ "github.com/filecoin-project/lily/model/actors/init"
@@ -104,6 +105,7 @@ var Models = []interface{}{
 	(*fevm.FEVMTrace)(nil),
 	(*actordumps.FEVMActorDump)(nil),
 	(*actordumps.MinerActorDump)(nil),
+	(*builtinactor.BuiltInActorEvent)(nil),
 }
 
 var log = logging.Logger("lily/storage")

tasks/api.go

@@ -86,6 +86,7 @@ type DataSource interface {
 	ChainGetMessagesInTipset(ctx context.Context, tsk types.TipSetKey) ([]api.Message, error)
 	EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error)
 	StateListActors(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error)
+	GetActorEventsRaw(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)
 
 	SetIdRobustAddressMap(ctx context.Context, tsk types.TipSetKey) error
 	LookupRobustAddress(ctx context.Context, idAddr address.Address, tsk types.TipSetKey) (address.Address, error)