feat: events: Implement API for clients to subscribe to and consume built-in Actor events

[aarshkshah1992]

Checklist

• This is not brainstorming ideas. If you have an idea you'd like to discuss, please open a new discussion on the lotus forum and select the category as Ideas.
• I have a specific, actionable, and well motivated feature request to propose.

Lotus component

• lotus daemon - chain sync
• lotus fvm/fevm - Lotus FVM and FEVM interactions
• lotus miner/worker - sealing
• lotus miner - proving(WindowPoSt/WinningPoSt)
• lotus JSON-RPC API
• lotus message management (mpool)
• Other

What is the motivation behind this feature request? Is your feature request related to a problem? Please describe.

Once filecoin-project/builtin-actors#1464 lands, built-in Actors will start emitting important lifecycle events that clients can subscribe to for tracking sector and deal lifecycle transitions.

We need to implement an API in Lotus for Lotus clients to be able to subscribe to these built-in Actor events.

---

1. Implement a new API for users to be able to consume (FEVM + built-in Actor) events that does not aggregate events by the tipsets/blocks they are interested in

We can implement a new API for clients to subscribe to (FEVM + built-in Actor) events i.e. all events emitted by a Filecoin node.

This API will provide an "here are the events you are interested in" view of the world rather than a "here's a new tipset/block and it has the following events you are interested in" i.e. it will not aggregate events by tipsets/blocks in it's response.

Emitted events will still contain information about the tipset/message they were included in/caused by.

This API will not emit information about tipset changes, message execution, mpool updates etc and clients can continue using the existing EthEventAPI or the existing ChainNotify API for those use cases.

Data Structures

type Block struct {
        // what value codec does client want to match on ?
	Codec uint64
	// data associated with the "event key" 
	Data  []byte 
}

type EventFilter struct {
	// Matches events from one of these actors, or any actor if empty.
	Addresses []address.Address
	// Matches events with the specified key/values, or all events if empty.
	// If the `Blocks` slice is empty, matches on the key only.
	Fields    map[string][]Block
	// Epoch based filtering
}

type Event struct {
	Entries     []types.EventEntry
	EmitterAddr address.Address // f4 address of emitter
	Reverted    bool
	Height      abi.ChainEpoch
	TipSetKey   cid.Cid // tipset that contained the message	
	MsgCid      cid.Cid         // cid of message that produced event
	// Do we need Event and Message Indices here ?
}

type EventEntry struct {
	// The key of this event entry
	Key string

	// The event value's codec
	Codec uint64

	// The event value
	Value []byte
}

Event Subscription API

type EventsAPI interface {
     GetHistoricalEvents(filter EventFilter) ([]Event, error)
     
     // TODO Do we need a web socket connection here ? Where do we write the events to  ?
     Subscribe(filter EventFilter) (<-chan []Event, error)     
}

---

2. Build a new API that allows clients to subscribe to (FEVM + built-in Actor events) aggregated by the blocks/tipsets they occur in

This API aggregates events by blocks/tipsets they are included in so clients can track event emission within the context of tipset changes.

This will need more development work as the existing event filter and Indexer does not already aggregate events by the tipsets they occur in.

Similar to #2 above with the following additions:

type BlockFilter struct {
	// Matches blocks from one of these miners, or any miner if empty.
	Miners []address.Address
}

type SubscriptionRequest struct {
	// Return blocks matching at least one of these filters, or blocks containing
	// messages/events matching one of the event/message filters.
	BlockFilter   []BlockFilter
	// Only return events matching at least one of these filters.
	EventFilter   []EventFilter
}
type HeadChange struct {
	// One of: "apply", "revert", or "current"
	Type   string
	TipSetKey cid.Cid
	Height    abi.ChainEpoch

	SubscribedEvents   []types.Event
}

type NotifyAPI interface {
     GetHistorical(filter SubscriptionRequest) ([]HeadChange, error)
     
     // TODO Do we need a web socket connection here ? Where do we write the events to  ?
     Subscribe(filter SubscriptionRequest) (<-chan []HeadChange, error)     
}

---

[aarshkshah1992]

@Stebalien Is there an easy way to hook into message execution so we can collect and emit executed messages that match the MessageFilter filter defined in #3 above ?

[Stebalien]

@Stebalien Is there an easy way to hook into message execution so we can collect and emit executed messages that match the MessageFilter filter defined in #3 above ?

I believe we discussed this sync but leaving a note here:

1. We can punt this for now.
2. We'd use a lotus event observer (to watch for chain changes), same as we do when indexing events.