feat: events: Lotus API to consume events (smart contract + built-in Actor events)

.circleci/config.yml

@@ -542,6 +542,12 @@ workflows:
       - docs-check:
           requires:
             - build
+      - test:
+          name: test-itest-actor_events_filter
+          requires:
+            - build
+          suite: itest-actor_events_filter
+          target: "./itests/actor_events_filter_test.go"
       - test:
           name: test-itest-api
           requires:

api/api_full.go

@@ -885,6 +885,22 @@ type FullNode interface {
 
 	RaftState(ctx context.Context) (*RaftStateData, error) //perm:read
 	RaftLeader(ctx context.Context) (peer.ID, error)       //perm:read
+
+	// Actor events
+
+	// GetActorEvents returns all FEVM Actor events that match the given filter.
+	// This is a request/response API. Please see the docs for `ActorEventFilter` for a detailed
+	// description of the filter options.
+	GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) //perm:read
+
+	// SubscribeActorEvents returns a long-lived stream of all FEVM Actor events that match the given filter.
+	// Events that match the given filter are written to the stream in real-time as they are emitted from the FEVM.
+	// The response stream is closed when the client disconnects or if there is an error while writing an event to the stream.
+	// This API also allows clients to read all historical events matching the given filter before
+	// any real-time events are written to the response stream.
+	// Please see the docs for `SubActorEventFilter` for a detailed description of the filter options.
+	// NOTE: THIS API IS ONLY SUPPORTED OVER WEBSOCKETS FOR NOW
+	SubscribeActorEvents(ctx context.Context, filter *types.SubActorEventFilter) (<-chan *types.ActorEvent, error) //perm:read
 }
 
 // reverse interface to the client, called after EthSubscribe

api/api_gateway.go

@@ -129,4 +129,7 @@ type Gateway interface {
 	Web3ClientVersion(ctx context.Context) (string, error)
 	EthTraceBlock(ctx context.Context, blkNum string) ([]*ethtypes.EthTraceBlock, error)
 	EthTraceReplayBlockTransactions(ctx context.Context, blkNum string, traceTypes []string) ([]*ethtypes.EthTraceReplayBlockTransaction, error)
+
+	GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)
+	SubscribeActorEvents(ctx context.Context, filter *types.SubActorEventFilter) (<-chan *types.ActorEvent, error)
 }

api/docgen/docgen.go

@@ -414,6 +414,42 @@ func init() {
 		VerifiedAllocationKey: nil,
 		Notify:                nil,
 	})
+
+	addExample(&types.ActorEventBlock{
+		Codec: 0x51,
+		Value: []byte("data"),
+	})
+
+	addExample(&types.ActorEventFilter{
+		Addresses: []address.Address{addr},
+		Fields: map[string][]types.ActorEventBlock{
+			"abc": {
+				{
+					Codec: 0x51,
+					Value: []byte("ddata"),
+				},
+			},
+		},
+		FromBlock: pstring("2301220"),
+		ToBlock:   pstring("latest"),
+	})
+
+	addExample(&types.SubActorEventFilter{
+		Filter: types.ActorEventFilter{
+			Addresses: []address.Address{addr},
+			Fields: map[string][]types.ActorEventBlock{
+				"abc": {
+					{
+						Codec: 0x51,
+						Value: []byte("data"),
+					},
+				},
+			},
+			FromBlock: pstring("2301220"),
+			ToBlock:   pstring("latest"),
+		},
+		Prefill: true,
+	})
 }
 
 func GetAPIType(name, pkg string) (i interface{}, t reflect.Type, permStruct []reflect.Type) {

chain/events/filter/event.go

@@ -28,13 +28,14 @@ func isIndexedValue(b uint8) bool {
 }
 
 type EventFilter struct {
-	id         types.FilterID
-	minHeight  abi.ChainEpoch // minimum epoch to apply filter or -1 if no minimum
-	maxHeight  abi.ChainEpoch // maximum epoch to apply filter or -1 if no maximum
-	tipsetCid  cid.Cid
-	addresses  []address.Address   // list of f4 actor addresses that are extpected to emit the event
-	keys       map[string][][]byte // map of key names to a list of alternate values that may match
-	maxResults int                 // maximum number of results to collect, 0 is unlimited
+	id        types.FilterID
+	minHeight abi.ChainEpoch // minimum epoch to apply filter or -1 if no minimum
+	maxHeight abi.ChainEpoch // maximum epoch to apply filter or -1 if no maximum
+	tipsetCid cid.Cid
+	addresses []address.Address // list of f4 actor addresses that are extpected to emit the event
+
+	keysWithCodec map[string][]types.ActorEventBlock // map of key names to a list of alternate values that may match
+	maxResults    int                                // maximum number of results to collect, 0 is unlimited
 
 	mu        sync.Mutex
 	collected []*CollectedEvent
@@ -194,7 +195,7 @@ func (f *EventFilter) matchAddress(o address.Address) bool {
 }
 
 func (f *EventFilter) matchKeys(ees []types.EventEntry) bool {
-	if len(f.keys) == 0 {
+	if len(f.keysWithCodec) == 0 {
 		return true
 	}
 	// TODO: optimize this naive algorithm
@@ -216,19 +217,19 @@ func (f *EventFilter) matchKeys(ees []types.EventEntry) bool {
 			continue
 		}
 
-		wantlist, ok := f.keys[keyname]
+		wantlist, ok := f.keysWithCodec[keyname]
 		if !ok || len(wantlist) == 0 {
 			continue
 		}
 
 		for _, w := range wantlist {
-			if bytes.Equal(w, ee.Value) {
+			if bytes.Equal(w.Value, ee.Value) && w.Codec == ee.Codec {
 				matched[keyname] = true
 				break
 			}
 		}
 
-		if len(matched) == len(f.keys) {
+		if len(matched) == len(f.keysWithCodec) {
 			// all keys have been matched
 			return true
 		}
@@ -362,7 +363,8 @@ func (m *EventFilterManager) Revert(ctx context.Context, from, to *types.TipSet)
 	return nil
 }
 
-func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight abi.ChainEpoch, tipsetCid cid.Cid, addresses []address.Address, keys map[string][][]byte) (*EventFilter, error) {
+func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight abi.ChainEpoch, tipsetCid cid.Cid, addresses []address.Address,
+	keysWithCodec map[string][]types.ActorEventBlock, excludeReverted bool) (*EventFilter, error) {
 	m.mu.Lock()
 	currentHeight := m.currentHeight
 	m.mu.Unlock()
@@ -377,18 +379,18 @@ func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight a
 	}
 
 	f := &EventFilter{
-		id:         id,
-		minHeight:  minHeight,
-		maxHeight:  maxHeight,
-		tipsetCid:  tipsetCid,
-		addresses:  addresses,
-		keys:       keys,
-		maxResults: m.MaxFilterResults,
+		id:            id,
+		minHeight:     minHeight,
+		maxHeight:     maxHeight,
+		tipsetCid:     tipsetCid,
+		addresses:     addresses,
+		keysWithCodec: keysWithCodec,
+		maxResults:    m.MaxFilterResults,
 	}
 
 	if m.EventIndex != nil && minHeight != -1 && minHeight < currentHeight {
 		// Filter needs historic events
-		if err := m.EventIndex.PrefillFilter(ctx, f, true); err != nil {
+		if err := m.EventIndex.PrefillFilter(ctx, f, excludeReverted); err != nil {
 			return nil, err
 		}
 	}