feat: ethereum filters over indexed events

chain/events/filter/event.go

@@ -18,10 +18,6 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 )
 
-type RobustAddresser interface {
-	LookupRobustAddress(ctx context.Context, idAddr address.Address, ts *types.TipSet) (address.Address, error)
-}
-
 const indexed uint8 = 0x01
 
 type EventFilter struct {
@@ -42,7 +38,7 @@ type EventFilter struct {
 var _ Filter = (*EventFilter)(nil)
 
 type CollectedEvent struct {
-	Event       *types.Event
+	Entries     []types.EventEntry
 	EmitterAddr address.Address // f4 address of emitter
 	EventIdx    int             // index of the event within the list of emitted events
 	Reverted    bool
@@ -104,7 +100,7 @@ func (f *EventFilter) CollectEvents(ctx context.Context, te *TipSetEvents, rever
 
 			// event matches filter, so record it
 			cev := &CollectedEvent{
-				Event:       ev,
+				Entries:     ev.Entries,
 				EmitterAddr: addr,
 				EventIdx:    evIdx,
 				Reverted:    revert,
@@ -134,6 +130,12 @@ func (f *EventFilter) CollectEvents(ctx context.Context, te *TipSetEvents, rever
 	return nil
 }
 
+func (f *EventFilter) setCollectedEvents(ces []*CollectedEvent) {
+	f.mu.Lock()
+	f.collected = ces
+	f.mu.Unlock()
+}
+
 func (f *EventFilter) TakeCollectedEvents(ctx context.Context) []*CollectedEvent {
 	f.mu.Lock()
 	collected := f.collected
@@ -198,7 +200,7 @@ func (f *EventFilter) matchKeys(ees []types.EventEntry) bool {
 			continue
 		}
 
-		keyname := string(ee.Key)
+		keyname := ee.Key
 
 		// skip if we have already matched this key
 		if matched[keyname] {
@@ -282,14 +284,18 @@ type EventFilterManager struct {
 	ChainStore       *cstore.ChainStore
 	AddressResolver  func(ctx context.Context, emitter abi.ActorID, ts *types.TipSet) (address.Address, bool)
 	MaxFilterResults int
+	EventIndex       *EventIndex
 
-	mu      sync.Mutex // guards mutations to filters
-	filters map[string]*EventFilter
+	mu            sync.Mutex // guards mutations to filters
+	filters       map[string]*EventFilter
+	currentHeight abi.ChainEpoch
 }
 
 func (m *EventFilterManager) Apply(ctx context.Context, from, to *types.TipSet) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	m.currentHeight = to.Height()
+
 	if len(m.filters) == 0 {
 		return nil
 	}
@@ -300,6 +306,12 @@ func (m *EventFilterManager) Apply(ctx context.Context, from, to *types.TipSet)
 		load:  m.loadExecutedMessages,
 	}
 
+	if m.EventIndex != nil {
+		if err := m.EventIndex.CollectEvents(ctx, tse, false, m.AddressResolver); err != nil {
+			return err
+		}
+	}
+
 	// TODO: could run this loop in parallel with errgroup if there are many filters
 	for _, f := range m.filters {
 		if err := f.CollectEvents(ctx, tse, false, m.AddressResolver); err != nil {
@@ -313,6 +325,8 @@ func (m *EventFilterManager) Apply(ctx context.Context, from, to *types.TipSet)
 func (m *EventFilterManager) Revert(ctx context.Context, from, to *types.TipSet) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	m.currentHeight = to.Height()
+
 	if len(m.filters) == 0 {
 		return nil
 	}
@@ -323,6 +337,12 @@ func (m *EventFilterManager) Revert(ctx context.Context, from, to *types.TipSet)
 		load:  m.loadExecutedMessages,
 	}
 
+	if m.EventIndex != nil {
+		if err := m.EventIndex.CollectEvents(ctx, tse, true, m.AddressResolver); err != nil {
+			return err
+		}
+	}
+
 	// TODO: could run this loop in parallel with errgroup if there are many filters
 	for _, f := range m.filters {
 		if err := f.CollectEvents(ctx, tse, true, m.AddressResolver); err != nil {
@@ -334,6 +354,14 @@ func (m *EventFilterManager) Revert(ctx context.Context, from, to *types.TipSet)
 }
 
 func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight abi.ChainEpoch, tipsetCid cid.Cid, addresses []address.Address, keys map[string][][]byte) (*EventFilter, error) {
+	m.mu.Lock()
+	currentHeight := m.currentHeight
+	m.mu.Unlock()
+
+	if m.EventIndex == nil && minHeight < currentHeight {
+		return nil, xerrors.Errorf("historic event index disabled")
+	}
+
 	id, err := uuid.NewRandom()
 	if err != nil {
 		return nil, xerrors.Errorf("new uuid: %w", err)
@@ -349,6 +377,13 @@ func (m *EventFilterManager) Install(ctx context.Context, minHeight, maxHeight a
 		maxResults: m.MaxFilterResults,
 	}
 
+	if m.EventIndex != nil && minHeight < currentHeight {
+		// Filter needs historic events
+		if err := m.EventIndex.PrefillFilter(ctx, f); err != nil {
+			return nil, err
+		}
+	}
+
 	m.mu.Lock()
 	m.filters[id.String()] = f
 	m.mu.Unlock()

chain/events/filter/event_test.go

@@ -60,7 +60,7 @@ func TestEventFilterCollectEvents(t *testing.T) {
 	noCollectedEvents := []*CollectedEvent{}
 	oneCollectedEvent := []*CollectedEvent{
 		{
-			Event:       ev1,
+			Entries:     ev1.Entries,
 			EmitterAddr: a1,
 			EventIdx:    0,
 			Reverted:    false,
@@ -286,15 +286,15 @@ func fakeEvent(emitter abi.ActorID, indexed []kv, unindexed []kv) *types.Event {
 	for _, in := range indexed {
 		ev.Entries = append(ev.Entries, types.EventEntry{
 			Flags: 0x01,
-			Key:   []byte(in.k),
+			Key:   in.k,
 			Value: in.v,
 		})
 	}
 
 	for _, in := range unindexed {
 		ev.Entries = append(ev.Entries, types.EventEntry{
 			Flags: 0x00,
-			Key:   []byte(in.k),
+			Key:   in.k,
 			Value: in.v,
 		})
 	}