feat: events: Add Lotus APIs to consume smart contract and built-in actor events

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 # UNRELEASED
 
+## New features
+
+### GetActorEvents and SubscribeActorEvents
+
+[FIP-0049](https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0049.md) introduced _Actor Events_ that can be emitted by user programmed actors. [FIP-0083](https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0083.md) introduces new events emitted by the builtin Verified Registry, Miner and Market Actors. These new events for builtin actors are being activated with network version 22 to coincide with _Direct Data Onboarding_ as defined in [FIP-0076](https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0076.md) which introduces additional flexibility for data onboarding. Sector, Deal and DataCap lifecycles can be tracked with these events, providing visibility and options for programmatic responses to changes in state.
+
+Actor events are available on message receipts, but can now be retrieved from a node using the new `GetActorEvents` and `SubscribeActorEvents` methods. These methods allow for querying and subscribing to actor events, respectively. They depend on the Lotus node both collecting events (with `Fevm.Events.RealTimeFilterAPI` and `Fevm.Events.HistoricFilterAPI`) and being enabled with the new configuration option `Events.EnableActorEventsAPI`. Note that a Lotus node can only respond to requests for historic events that it retains in its event store.
+
+Both `GetActorEvents` and `SubscribeActorEvents` take a filter parameter which can optionally filter events on:
+
+* `Addresses` of the actor(s) emitting the event
+* Specific `Fields` within the event
+* `FromHeight` and `ToHeight` to filter events by block height
+* `TipSetKey` to restrict events contained within a specific tipset
+
+`GetActorEvents` provides a one-time query for actor events, while `SubscribeActorEvents` provides a long-lived connection (via websockets) to the Lotus node, allowing for real-time updates on actor events. The subscription can be cancelled by the client at any time.
+
+### GetAllClaims and GetAllAlocations
+
+2 new methods have benn added to the Lotus API called `GetAllAllocations` and `GetAllClaims` which lists all the available allocations and claims available in the actor state.
+
+See [FIP-0045](https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0045.md) for background on claims and allocations.
+
 ## Improvements
 
 ### Tracing API
@@ -49,10 +72,8 @@ Replace the `CodeCid` field in the message trace (added in 1.23.4) with an `Invo
 ```
 This means the trace now contains an accurate "snapshot" of the actor at the time of the call, information that may not be present in the final state-tree (e.g., due to reverts). This will hopefully improve the performance and accuracy of indexing services.
 
-### Lotus API
-2 new methods have benn added to the Lotus API called `GetAllAllocations` and `GetAllClaims` which lists all the available allocations and claims available in the actor state.
-
 ### Lotus CLI
+
 The `filplus` commands used for listing allocations and claims have been updated. If no argument is provided to the either command, they will list out all the allocations and claims in the verified registry actor.
 The output list columns have been modified to `AllocationID` and `ClaimID` instead of ID. 
 

api/api_full.go

@@ -906,6 +906,28 @@ 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 FVM and built-in Actor events that match the given filter.
+	// This is a request/response API.
+	// Results available from this API may be limited by the MaxFilterResults and MaxFilterHeightRange
+	// configuration options and also the amount of historical data available in the node.
+	GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) //perm:read
+
+	// SubscribeActorEvents returns a long-lived stream of all FVM and built-in 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 FVM.
+	// The response stream is closed when the client disconnects, when a ToHeight is specified and is
+	// reached, 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 if the filter specifies an earlier
+	// FromHeight.
+	// Results available from this API may be limited by the MaxFilterResults and MaxFilterHeightRange
+	// configuration options and also the amount of historical data available in the node.
+	// NOTE: THIS API IS ONLY SUPPORTED OVER WEBSOCKETS FOR NOW
+	SubscribeActorEvents(ctx context.Context, filter *types.ActorEventFilter) (<-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.ActorEventFilter) (<-chan *types.ActorEvent, error)
 }

api/docgen/docgen.go

@@ -414,6 +414,25 @@ func init() {
 		VerifiedAllocationKey: nil,
 		Notify:                nil,
 	})
+
+	addExample(&types.ActorEventBlock{
+		Codec: 0x51,
+		Value: []byte("ddata"),
+	})
+
+	addExample(&types.ActorEventFilter{
+		Addresses: []address.Address{addr},
+		Fields: map[string][]types.ActorEventBlock{
+			"abc": {
+				{
+					Codec: 0x51,
+					Value: []byte("ddata"),
+				},
+			},
+		},
+		FromHeight: epochPtr(1010),
+		ToHeight:   epochPtr(1020),
+	})
 }
 
 func GetAPIType(name, pkg string) (i interface{}, t reflect.Type, permStruct []reflect.Type) {
@@ -519,6 +538,11 @@ func exampleStruct(method string, t, parent reflect.Type) interface{} {
 	return ns.Interface()
 }
 
+func epochPtr(ei int64) *abi.ChainEpoch {
+	ep := abi.ChainEpoch(ei)
+	return &ep
+}
+
 type Visitor struct {
 	Root    string
 	Methods map[string]ast.Node

chain/events/cache.go

@@ -26,7 +26,7 @@ type cache struct {
 	uncachedAPI
 }
 
-func newCache(api EventAPI, gcConfidence abi.ChainEpoch) *cache {
+func newCache(api EventHelperAPI, gcConfidence abi.ChainEpoch) *cache {
 	return &cache{
 		newTSCache(api, gcConfidence),
 		newMessageCache(api),

chain/events/events.go

@@ -28,7 +28,7 @@ type TipSetObserver interface {
 	Revert(ctx context.Context, from, to *types.TipSet) error
 }
 
-type EventAPI interface {
+type EventHelperAPI interface {
 	ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
 	ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
 	ChainGetTipSetByHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
@@ -47,7 +47,7 @@ type Events struct {
 	*hcEvents
 }
 
-func newEventsWithGCConfidence(ctx context.Context, api EventAPI, gcConfidence abi.ChainEpoch) (*Events, error) {
+func newEventsWithGCConfidence(ctx context.Context, api EventHelperAPI, gcConfidence abi.ChainEpoch) (*Events, error) {
 	cache := newCache(api, gcConfidence)
 
 	ob := newObserver(cache, gcConfidence)
@@ -61,7 +61,7 @@ func newEventsWithGCConfidence(ctx context.Context, api EventAPI, gcConfidence a
 	return &Events{ob, he, headChange}, nil
 }
 
-func NewEvents(ctx context.Context, api EventAPI) (*Events, error) {
+func NewEvents(ctx context.Context, api EventHelperAPI) (*Events, error) {
 	gcConfidence := 2 * build.ForkLengthThreshold
 	return newEventsWithGCConfidence(ctx, api, gcConfidence)
 }

chain/events/events_called.go

@@ -69,7 +69,7 @@ type queuedEvent struct {
 // Manages chain head change events, which may be forward (new tipset added to
 // chain) or backward (chain branch discarded in favour of heavier branch)
 type hcEvents struct {
-	cs EventAPI
+	cs EventHelperAPI
 
 	lk     sync.Mutex
 	lastTs *types.TipSet
@@ -94,7 +94,7 @@ type hcEvents struct {
 	watcherEvents
 }
 
-func newHCEvents(api EventAPI, obs *observer) *hcEvents {
+func newHCEvents(api EventHelperAPI, obs *observer) *hcEvents {
 	e := &hcEvents{
 		cs:          api,
 		confQueue:   map[triggerH]map[msgH][]*queuedEvent{},
@@ -326,14 +326,14 @@ type headChangeAPI interface {
 
 // watcherEvents watches for a state change
 type watcherEvents struct {
-	cs    EventAPI
+	cs    EventHelperAPI
 	hcAPI headChangeAPI
 
 	lk       sync.RWMutex
 	matchers map[triggerID]StateMatchFunc
 }
 
-func newWatcherEvents(hcAPI headChangeAPI, cs EventAPI) watcherEvents {
+func newWatcherEvents(hcAPI headChangeAPI, cs EventHelperAPI) watcherEvents {
 	return watcherEvents{
 		cs:       cs,
 		hcAPI:    hcAPI,
@@ -426,14 +426,14 @@ func (we *watcherEvents) StateChanged(check CheckFunc, scHnd StateChangeHandler,
 
 // messageEvents watches for message calls to actors
 type messageEvents struct {
-	cs    EventAPI
+	cs    EventHelperAPI
 	hcAPI headChangeAPI
 
 	lk       sync.RWMutex
 	matchers map[triggerID]MsgMatchFunc
 }
 
-func newMessageEvents(hcAPI headChangeAPI, cs EventAPI) messageEvents {
+func newMessageEvents(hcAPI headChangeAPI, cs EventHelperAPI) messageEvents {
 	return messageEvents{
 		cs:       cs,
 		hcAPI:    hcAPI,

chain/events/events_height.go

@@ -22,7 +22,7 @@ type heightHandler struct {
 }
 
 type heightEvents struct {
-	api          EventAPI
+	api          EventHelperAPI
 	gcConfidence abi.ChainEpoch
 
 	lk                        sync.Mutex
@@ -31,7 +31,7 @@ type heightEvents struct {
 	lastGc                    abi.ChainEpoch //nolint:structcheck
 }
 
-func newHeightEvents(api EventAPI, obs *observer, gcConfidence abi.ChainEpoch) *heightEvents {
+func newHeightEvents(api EventHelperAPI, obs *observer, gcConfidence abi.ChainEpoch) *heightEvents {
 	he := &heightEvents{
 		api:            api,
 		gcConfidence:   gcConfidence,

chain/events/events_test.go

@@ -358,7 +358,7 @@ func (fcs *fakeCS) advance(rev, app, drop int, msgs map[int]cid.Cid, nulls ...in
 	fcs.sub(nil, nil)
 }
 
-var _ EventAPI = &fakeCS{}
+var _ EventHelperAPI = &fakeCS{}
 
 func TestAt(t *testing.T) {
 	//stm: @EVENTS_HEIGHT_CHAIN_AT_001, @EVENTS_HEIGHT_REVERT_001