lotus/node/impl/full/actor_events.go

package full

import (
	"context"
	"fmt"
	"time"

	"github.com/ipfs/go-cid"
	"go.uber.org/fx"

	"github.com/filecoin-project/go-state-types/abi"

	"github.com/filecoin-project/lotus/api"
	"github.com/filecoin-project/lotus/build"
	"github.com/filecoin-project/lotus/chain/events/filter"
	"github.com/filecoin-project/lotus/chain/store"
	"github.com/filecoin-project/lotus/chain/types"
)

type ActorEventAPI interface {
	GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)
	SubscribeActorEvents(ctx context.Context, filter *types.ActorEventFilter) (<-chan *types.ActorEvent, error)
}

var (
	_ ActorEventAPI = *new(api.FullNode)
	_ ActorEventAPI = *new(api.Gateway)
)

type ActorEventHandler struct {
	EventFilterManager   *filter.EventFilterManager
	MaxFilterHeightRange abi.ChainEpoch
	Chain                *store.ChainStore
}

var _ ActorEventAPI = (*ActorEventHandler)(nil)

type ActorEventsAPI struct {
	fx.In
	ActorEventAPI
}

func (a *ActorEventHandler) GetActorEvents(ctx context.Context, evtFilter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
	if a.EventFilterManager == nil {
		return nil, api.ErrNotSupported
	}

	if evtFilter == nil {
		evtFilter = &types.ActorEventFilter{}
	}
	params, err := a.parseFilter(*evtFilter)
	if err != nil {
		return nil, err
	}

	// Install a filter just for this call, collect events, remove the filter

	tipSetCid, err := params.GetTipSetCid()
	if err != nil {
		return nil, fmt.Errorf("failed to get tipset cid: %w", err)
	}
	f, err := a.EventFilterManager.Install(ctx, params.MinHeight, params.MaxHeight, tipSetCid, evtFilter.Addresses, evtFilter.Fields, false)
	if err != nil {
		return nil, err
	}

	evs, _, _ := getCollected(ctx, f)
	if err := a.EventFilterManager.Remove(ctx, f.ID()); err != nil {
		log.Warnf("failed to remove filter: %s", err)
	}
	return evs, nil
}

type filterParams struct {
	MinHeight abi.ChainEpoch
	MaxHeight abi.ChainEpoch
	TipSetKey types.TipSetKey
}

func (fp filterParams) GetTipSetCid() (cid.Cid, error) {
	if fp.TipSetKey.IsEmpty() {
		return cid.Undef, nil
	}
	return fp.TipSetKey.Cid()
}

func (a *ActorEventHandler) parseFilter(f types.ActorEventFilter) (*filterParams, error) {
	if f.TipSetKey != nil && !f.TipSetKey.IsEmpty() {
		if f.FromHeight != nil || f.ToHeight != nil {
			return nil, fmt.Errorf("cannot specify both TipSetKey and FromHeight/ToHeight")
		}

		tsk := types.EmptyTSK
		if f.TipSetKey != nil {
			tsk = *f.TipSetKey
		}
		return &filterParams{
			MinHeight: 0,
			MaxHeight: 0,
			TipSetKey: tsk,
		}, nil
	}

	min, max, err := parseHeightRange(a.Chain.GetHeaviestTipSet().Height(), f.FromHeight, f.ToHeight, a.MaxFilterHeightRange)
	if err != nil {
		return nil, err
	}

	return &filterParams{
		MinHeight: min,
		MaxHeight: max,
		TipSetKey: types.EmptyTSK,
	}, nil
}

// parseHeightRange is similar to eth's parseBlockRange but with slightly different semantics but
// results in equivalent values that we can plug in to the EventFilterManager.
//
// * Uses "height", allowing for nillable values rather than strings
// * No "latest" and "earliest", those are now represented by nil on the way in and -1 on the way out
// * No option for hex representation
func parseHeightRange(heaviest abi.ChainEpoch, fromHeight, toHeight *abi.ChainEpoch, maxRange abi.ChainEpoch) (minHeight abi.ChainEpoch, maxHeight abi.ChainEpoch, err error) {
	if fromHeight != nil && *fromHeight < 0 {
		return 0, 0, fmt.Errorf("range 'from' must be greater than or equal to 0")
	}
	if fromHeight == nil {
		minHeight = -1
	} else {
		minHeight = *fromHeight
	}
	if toHeight == nil {
		maxHeight = -1
	} else {
		maxHeight = *toHeight
	}

	// Validate height ranges are within limits set by node operator
	if minHeight == -1 && maxHeight > 0 {
		// Here the client is looking for events between the head and some future height
		if maxHeight-heaviest > maxRange {
			return 0, 0, fmt.Errorf("invalid epoch range: 'to' height is too far in the future (maximum: %d)", maxRange)
		}
	} else if minHeight >= 0 && maxHeight == -1 {
		// Here the client is looking for events between some time in the past and the current head
		if heaviest-minHeight > maxRange {
			return 0, 0, fmt.Errorf("invalid epoch range: 'from' height is too far in the past (maximum: %d)", maxRange)
		}
	} else if minHeight >= 0 && maxHeight >= 0 {
		if minHeight > maxHeight {
			return 0, 0, fmt.Errorf("invalid epoch range: 'to' height (%d) must be after 'from' height (%d)", minHeight, maxHeight)
		} else if maxHeight-minHeight > maxRange {
			return 0, 0, fmt.Errorf("invalid epoch range: range between to and 'from' heights is too large (maximum: %d)", maxRange)
		}
	}
	return minHeight, maxHeight, nil
}

func (a *ActorEventHandler) SubscribeActorEvents(ctx context.Context, evtFilter *types.ActorEventFilter) (<-chan *types.ActorEvent, error) {
	if a.EventFilterManager == nil {
		return nil, api.ErrNotSupported
	}
	if evtFilter == nil {
		evtFilter = &types.ActorEventFilter{}
	}
	params, err := a.parseFilter(*evtFilter)
	if err != nil {
		return nil, err
	}

	tipSetCid, err := params.GetTipSetCid()
	if err != nil {
		return nil, fmt.Errorf("failed to get tipset cid: %w", err)
	}
	fm, err := a.EventFilterManager.Install(ctx, params.MinHeight, params.MaxHeight, tipSetCid, evtFilter.Addresses, evtFilter.Fields, false)
	if err != nil {
		return nil, err
	}

	// The goal for the code below is to be able to send events on the `out` channel as fast as
	// possible and not let it get too far behind the rate at which the events are generated.
	// For historical events we see the rate at which they were generated by looking the height range;
	// we then make sure that the client can receive them at least twice as fast as they were
	// generated so they catch up quick enough to receive new events.
	// For ongoing events we use an exponential moving average of the events per height to make sure
	// that the client doesn't fall behind.
	// In both cases we allow a little bit of slack but need to avoid letting the client bloat the
	// buffer too much.
	// There is no special handling for reverts, so they will just look like a lot more events per
	// epoch and the user has to receive them anyway.

	out := make(chan *types.ActorEvent)

	go func() {
		defer func() {
			// tell the caller we're done
			close(out)
			fm.ClearSubChannel()
			if err := a.EventFilterManager.Remove(ctx, fm.ID()); err != nil {
				log.Warnf("failed to remove filter: %s", err)
			}
		}()

		// Handle any historical events that our filter may have picked up -----------------------------

		evs, minEpoch, maxEpoch := getCollected(ctx, fm)
		if len(evs) > 0 {
			// must be able to send events at least twice as fast as they were generated
			epochRange := maxEpoch - minEpoch
			if epochRange <= 0 {
				epochRange = 1
			}
			eventsPerEpoch := float64(len(evs)) / float64(epochRange)
			eventsPerSecond := 2 * eventsPerEpoch / float64(build.BlockDelaySecs)
			// a minimum rate of 1 event per second if we don't have many events
			if eventsPerSecond < 1 {
				eventsPerSecond = 1
			}

			// send events from evs to the out channel and ensure we don't do it slower than eventsPerMs
			ticker := time.NewTicker(time.Second)
			defer ticker.Stop()

			const maxSlowTicks = 3 // slightly forgiving, allow 3 slow ticks (seconds) before giving up
			slowTicks := 0
			sentEvents := 0.0

			for _, ev := range evs {
				select {
				case out <- ev:
					sentEvents++
				case <-ticker.C:
					if sentEvents < eventsPerSecond {
						slowTicks++
						if slowTicks >= maxSlowTicks {
							log.Errorf("closing event subscription due to slow event sending rate")
							return
						}
					} else {
						slowTicks = 0
					}
					sentEvents = 0
				case <-ctx.Done():
					return
				}
			}
		}

		// Handle ongoing events from the filter -------------------------------------------------------

		in := make(chan interface{}, 256)
		fm.SetSubChannel(in)

		var buffer []*types.ActorEvent
		const α = 0.2                        // decay factor for the events per height EMA
		var eventsPerHeightEma float64 = 256 // exponential moving average of events per height, initially guess at 256
		var lastHeight abi.ChainEpoch        // last seen event height
		var eventsAtCurrentHeight int        // number of events at the current height

		collectEvent := func(ev interface{}) bool {
			ce, ok := ev.(*filter.CollectedEvent)
			if !ok {
				log.Errorf("got unexpected value from event filter: %T", ev)
				return false
			}

			if ce.Height > lastHeight {
				// update the EMA of events per height when the height increases
				if lastHeight != 0 {
					eventsPerHeightEma = α*float64(eventsAtCurrentHeight) + (1-α)*eventsPerHeightEma
				}
				lastHeight = ce.Height
				eventsAtCurrentHeight = 0
			}
			eventsAtCurrentHeight++

			buffer = append(buffer, &types.ActorEvent{
				Entries:   ce.Entries,
				Emitter:   ce.EmitterAddr,
				Reverted:  ce.Reverted,
				Height:    ce.Height,
				TipSetKey: ce.TipSetKey,
				MsgCid:    ce.MsgCid,
			})
			return true
		}

		// for the case where we have a MaxHeight set, we don't get a signal from the filter when we
		// reach that height, so we need to check it ourselves, do it now but also in the loop
		if params.MaxHeight > 0 && a.Chain.GetHeaviestTipSet().Height() > params.MaxHeight {
			return
		}
		ticker := time.NewTicker(time.Duration(build.BlockDelaySecs) * time.Second)
		defer ticker.Stop()

		for ctx.Err() == nil {
			if len(buffer) > 0 {
				// check if we need to disconnect the client because they've fallen behind, always allow at
				// least 8 events in the buffer to provide a little bit of slack
				if len(buffer) > 8 && float64(len(buffer)) > eventsPerHeightEma/2 {
					log.Errorf("closing event subscription due to slow event sending rate")
					return
				}

				select {
				case ev, ok := <-in: // incoming event
					if !ok || !collectEvent(ev) {
						return
					}
				case out <- buffer[0]: // successful send
					buffer[0] = nil
					buffer = buffer[1:]
				case <-ctx.Done():
					return
				}
			} else {
				select {
				case ev, ok := <-in: // incoming event
					if !ok || !collectEvent(ev) {
						return
					}
				case <-ctx.Done():
					return
				case <-ticker.C:
					if params.MaxHeight > 0 && a.Chain.GetHeaviestTipSet().Height() > params.MaxHeight {
						return
					}
				}
			}
		}
	}()

	return out, nil
}

func getCollected(ctx context.Context, f *filter.EventFilter) ([]*types.ActorEvent, abi.ChainEpoch, abi.ChainEpoch) {
	ces := f.TakeCollectedEvents(ctx)

	var out []*types.ActorEvent
	var min, max abi.ChainEpoch

	for _, e := range ces {
		if min == 0 || e.Height < min {
			min = e.Height
		}
		if e.Height > max {
			max = e.Height
		}
		out = append(out, &types.ActorEvent{
			Entries:   e.Entries,
			Emitter:   e.EmitterAddr,
			Reverted:  e.Reverted,
			Height:    e.Height,
			TipSetKey: e.TipSetKey,
			MsgCid:    e.MsgCid,
		})
	}

	return out, min, max
}