Adjust actor event API after review

node/builder_chain.go

@@ -281,10 +281,10 @@ func ConfigFullNode(c interface{}) Option {
 		),
 
 		ApplyIf(isFullNode,
-			If(cfg.Fevm.EnableActorEventsAPI,
-				Override(new(full.ActorEventAPI), modules.ActorEventHandler(cfg.Fevm)),
+			If(cfg.ActorEvents.EnableActorEventsAPI,
+				Override(new(full.ActorEventAPI), modules.ActorEventHandler(cfg.ActorEvents.EnableActorEventsAPI, cfg.Fevm)),
 			),
-			If(!cfg.Fevm.EnableActorEventsAPI,
+			If(!cfg.ActorEvents.EnableActorEventsAPI,
 				Override(new(full.ActorEventAPI), &full.ActorEventDummy{}),
 			),
 		),

node/config/def.go

@@ -109,7 +109,6 @@ func DefaultFullNode() *FullNode {
 		Cluster: *DefaultUserRaftConfig(),
 		Fevm: FevmConfig{
 			EnableEthRPC:                 false,
-			EnableActorEventsAPI:         false,
 			EthTxHashMappingLifetimeDays: 0,
 			Events: Events{
 				DisableRealTimeFilterAPI: false,
@@ -120,6 +119,9 @@ func DefaultFullNode() *FullNode {
 				MaxFilterHeightRange:     2880, // conservative limit of one day
 			},
 		},
+		ActorEvents: ActorEventsConfig{
+			EnableActorEventsAPI: false,
+		},
 	}
 }
 

node/config/doc_gen.go

@@ -29,6 +29,17 @@ var Doc = map[string][]DocField{
 			Comment: ``,
 		},
 	},
+	"ActorEventsConfig": {
+		{
+			Name: "EnableActorEventsAPI",
+			Type: "bool",
+
+			Comment: `EnableActorEventsAPI enables the Actor events API that enables clients to consume events
+emitted by (smart contracts + built-in Actors).
+This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be
+disabled by setting their respective Disable* options in Fevm.Events.`,
+		},
+	},
 	"ApisConfig": {
 		{
 			Name: "ChainApiInfo",
@@ -452,13 +463,6 @@ rewards. This address should have adequate funds to cover gas fees.`,
 			Type: "bool",
 
 			Comment: `EnableEthRPC enables eth_ rpc, and enables storing a mapping of eth transaction hashes to filecoin message Cids.
-This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be disabled by config options above.`,
-		},
-		{
-			Name: "EnableActorEventsAPI",
-			Type: "bool",
-
-			Comment: `EnableActorEventsAPI enables the Actor events API that enables clients to consume events emitted by (smart contracts + built-in Actors).
 This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be disabled by config options above.`,
 		},
 		{
@@ -512,6 +516,12 @@ Set to 0 to keep all mappings`,
 
 			Comment: ``,
 		},
+		{
+			Name: "ActorEvents",
+			Type: "ActorEventsConfig",
+
+			Comment: ``,
+		},
 		{
 			Name: "Index",
 			Type: "IndexConfig",

node/config/types.go

@@ -28,6 +28,7 @@ type FullNode struct {
 	Chainstore    Chainstore
 	Cluster       UserRaftConfig
 	Fevm          FevmConfig
+	ActorEvents   ActorEventsConfig
 	Index         IndexConfig
 	FaultReporter FaultReporterConfig
 }
@@ -786,10 +787,6 @@ type FevmConfig struct {
 	// This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be disabled by config options above.
 	EnableEthRPC bool
 
-	// EnableActorEventsAPI enables the Actor events API that enables clients to consume events emitted by (smart contracts + built-in Actors).
-	// This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be disabled by config options above.
-	EnableActorEventsAPI bool
-
 	// EthTxHashMappingLifetimeDays the transaction hash lookup database will delete mappings that have been stored for more than x days
 	// Set to 0 to keep all mappings
 	EthTxHashMappingLifetimeDays int
@@ -833,6 +830,14 @@ type Events struct {
 	// Set upper bound on index size
 }
 
+type ActorEventsConfig struct {
+	// EnableActorEventsAPI enables the Actor events API that enables clients to consume events
+	// emitted by (smart contracts + built-in Actors).
+	// This will also enable the RealTimeFilterAPI and HistoricFilterAPI by default, but they can be
+	// disabled by setting their respective Disable* options in Fevm.Events.
+	EnableActorEventsAPI bool
+}
+
 type IndexConfig struct {
 	// EXPERIMENTAL FEATURE. USE WITH CAUTION
 	// EnableMsgIndex enables indexing of messages on chain.
@@ -856,6 +861,7 @@ type HarmonyDB struct {
 	// The port to find Yugabyte. Blank for default.
 	Port string
 }
+
 type FaultReporterConfig struct {
 	// EnableConsensusFaultReporter controls whether the node will monitor and
 	// report consensus faults. When enabled, the node will watch for malicious

node/impl/full/actor_event_test.go

@@ -0,0 +1,112 @@
+package full
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/go-state-types/abi"
+)
+
+func TestParseHeightRange(t *testing.T) {
+	epochPtr := func(i int) *abi.ChainEpoch {
+		e := abi.ChainEpoch(i)
+		return &e
+	}
+
+	tcs := map[string]struct {
+		heaviest abi.ChainEpoch
+		from     *abi.ChainEpoch
+		to       *abi.ChainEpoch
+		maxRange abi.ChainEpoch
+		minOut   abi.ChainEpoch
+		maxOut   abi.ChainEpoch
+		errStr   string
+	}{
+		"fails when both are specified and range is greater than max allowed range": {
+			heaviest: 100,
+			from:     epochPtr(256),
+			to:       epochPtr(512),
+			maxRange: 10,
+			minOut:   0,
+			maxOut:   0,
+			errStr:   "too large",
+		},
+		"fails when min is specified and range is greater than max allowed range": {
+			heaviest: 500,
+			from:     epochPtr(16),
+			to:       nil,
+			maxRange: 10,
+			minOut:   0,
+			maxOut:   0,
+			errStr:   "'from' height is too far in the past",
+		},
+		"fails when max is specified and range is greater than max allowed range": {
+			heaviest: 500,
+			from:     nil,
+			to:       epochPtr(65536),
+			maxRange: 10,
+			minOut:   0,
+			maxOut:   0,
+			errStr:   "'to' height is too far in the future",
+		},
+		"fails when from is greater than to": {
+			heaviest: 100,
+			from:     epochPtr(512),
+			to:       epochPtr(256),
+			maxRange: 10,
+			minOut:   0,
+			maxOut:   0,
+			errStr:   "must be after",
+		},
+		"works when range is valid (nil from)": {
+			heaviest: 500,
+			from:     nil,
+			to:       epochPtr(48),
+			maxRange: 1000,
+			minOut:   -1,
+			maxOut:   48,
+		},
+		"works when range is valid (nil to)": {
+			heaviest: 500,
+			from:     epochPtr(0),
+			to:       nil,
+			maxRange: 1000,
+			minOut:   0,
+			maxOut:   -1,
+		},
+		"works when range is valid (nil from and to)": {
+			heaviest: 500,
+			from:     nil,
+			to:       nil,
+			maxRange: 1000,
+			minOut:   -1,
+			maxOut:   -1,
+		},
+		"works when range is valid and specified": {
+			heaviest: 500,
+			from:     epochPtr(16),
+			to:       epochPtr(48),
+			maxRange: 1000,
+			minOut:   16,
+			maxOut:   48,
+		},
+	}
+
+	for name, tc := range tcs {
+		tc2 := tc
+		t.Run(name, func(t *testing.T) {
+			min, max, err := parseHeightRange(tc2.heaviest, tc2.from, tc2.to, tc2.maxRange)
+			require.Equal(t, tc2.minOut, min)
+			require.Equal(t, tc2.maxOut, max)
+			if tc2.errStr != "" {
+				fmt.Println(err)
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc2.errStr)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

node/impl/full/actor_events.go

@@ -3,7 +3,6 @@ package full
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/ipfs/go-cid"
 	"go.uber.org/fx"
@@ -18,7 +17,7 @@ import (
 
 type ActorEventAPI interface {
 	GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error)
-	SubscribeActorEvents(ctx context.Context, filter *types.SubActorEventFilter) (<-chan *types.ActorEvent, error)
+	SubscribeActorEvents(ctx context.Context, filter *types.ActorEventFilter) (<-chan *types.ActorEvent, error)
 }
 
 var (
@@ -39,18 +38,26 @@ type ActorEventsAPI struct {
 	ActorEventAPI
 }
 
-func (a *ActorEventHandler) GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+func (a *ActorEventHandler) GetActorEvents(ctx context.Context, evtFilter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
 	if a.EventFilterManager == nil {
 		return nil, api.ErrNotSupported
 	}
 
-	params, err := a.parseFilter(filter)
+	if evtFilter == nil {
+		evtFilter = &types.ActorEventFilter{}
+	}
+	params, err := a.parseFilter(*evtFilter)
 	if err != nil {
 		return nil, err
 	}
 
-	// Create a temporary filter
-	f, err := a.EventFilterManager.Install(ctx, params.MinHeight, params.MaxHeight, params.TipSetCid, filter.Addresses, filter.Fields, false)
+	// 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
 	}
@@ -63,33 +70,34 @@ func (a *ActorEventHandler) GetActorEvents(ctx context.Context, filter *types.Ac
 type filterParams struct {
 	MinHeight abi.ChainEpoch
 	MaxHeight abi.ChainEpoch
-	TipSetCid cid.Cid
+	TipSetKey types.TipSetKey
 }
 
-func (a *ActorEventHandler) parseFilter(f *types.ActorEventFilter) (*filterParams, error) {
-	if f.TipSetCid != nil {
-		if len(f.FromEpoch) != 0 || len(f.ToEpoch) != 0 {
-			return nil, fmt.Errorf("cannot specify both TipSetCid and FromEpoch/ToEpoch")
+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,
-			TipSetCid: *f.TipSetCid,
+			TipSetKey: tsk,
 		}, nil
 	}
 
-	from := f.FromEpoch
-	if len(from) != 0 && from != "latest" && from != "earliest" && !strings.HasPrefix(from, "0x") {
-		from = "0x" + from
-	}
-
-	to := f.ToEpoch
-	if len(to) != 0 && to != "latest" && to != "earliest" && !strings.HasPrefix(to, "0x") {
-		to = "0x" + to
-	}
-
-	min, max, err := parseBlockRange(a.Chain.GetHeaviestTipSet().Height(), &from, &to, a.MaxFilterHeightRange)
+	min, max, err := parseHeightRange(a.Chain.GetHeaviestTipSet().Height(), f.FromHeight, f.ToHeight, a.MaxFilterHeightRange)
 	if err != nil {
 		return nil, err
 	}
@@ -97,21 +105,69 @@ func (a *ActorEventHandler) parseFilter(f *types.ActorEventFilter) (*filterParam
 	return &filterParams{
 		MinHeight: min,
 		MaxHeight: max,
-		TipSetCid: cid.Undef,
+		TipSetKey: types.EmptyTSK,
 	}, nil
 }
 
-func (a *ActorEventHandler) SubscribeActorEvents(ctx context.Context, f *types.SubActorEventFilter) (<-chan *types.ActorEvent, error) {
+// 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
 	}
-
-	params, err := a.parseFilter(&f.Filter)
+	if evtFilter == nil {
+		evtFilter = &types.ActorEventFilter{}
+	}
+	params, err := a.parseFilter(*evtFilter)
 	if err != nil {
 		return nil, err
 	}
 
-	fm, err := a.EventFilterManager.Install(ctx, params.MinHeight, params.MaxHeight, params.TipSetCid, f.Filter.Addresses, f.Filter.Fields, false)
+	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
 	}
@@ -128,30 +184,29 @@ func (a *ActorEventHandler) SubscribeActorEvents(ctx context.Context, f *types.S
 			_ = a.EventFilterManager.Remove(ctx, fm.ID())
 		}()
 
-		if f.Prefill {
-			evs, err := getCollected(ctx, fm)
-			if err != nil {
-				log.Errorf("failed to get collected events: %w", err)
-				return
-			}
+		evs, err := getCollected(ctx, fm)
+		if err != nil {
+			log.Errorf("failed to get collected events: %w", err)
+			return
+		}
 
-			for _, ev := range evs {
-				ev := ev
-				select {
-				case out <- ev:
-				case <-ctx.Done():
-					return
-				default:
-					log.Errorf("closing event subscription due to slow reader")
-					return
-				}
+		for _, ev := range evs {
+			ev := ev
+			select {
+			case out <- ev:
+			case <-ctx.Done():
+				return
+			default:
+				// TODO: need to fix this, buffer of 25 isn't going to work for prefill without a _really_ fast client or a small number of events
+				log.Errorf("closing event subscription due to slow reader")
+				return
 			}
 		}
 
 		in := make(chan interface{}, 256)
 		fm.SetSubChannel(in)
 
-		for {
+		for ctx.Err() == nil {
 			select {
 			case val, ok := <-in:
 				if !ok {
@@ -164,24 +219,19 @@ func (a *ActorEventHandler) SubscribeActorEvents(ctx context.Context, f *types.S
 					log.Errorf("got unexpected value from event filter: %T", val)
 					return
 				}
-				c, err := ce.TipSetKey.Cid()
-				if err != nil {
-					log.Errorf("failed to get tipset cid: %w", err)
-					return
-				}
 
 				ev := &types.ActorEvent{
-					Entries:     ce.Entries,
-					EmitterAddr: ce.EmitterAddr,
-					Reverted:    ce.Reverted,
-					Height:      ce.Height,
-					TipSetCid:   c,
-					MsgCid:      ce.MsgCid,
+					Entries:   ce.Entries,
+					Emitter:   ce.EmitterAddr,
+					Reverted:  ce.Reverted,
+					Height:    ce.Height,
+					TipSetKey: ce.TipSetKey,
+					MsgCid:    ce.MsgCid,
 				}
 
 				select {
 				case out <- ev:
-				default:
+				default: // TODO: need to fix this to be more intelligent about the consumption rate vs the accumulation rate
 					log.Errorf("closing event subscription due to slow reader")
 					return
 				}
@@ -204,22 +254,14 @@ func getCollected(ctx context.Context, f *filter.EventFilter) ([]*types.ActorEve
 	var out []*types.ActorEvent
 
 	for _, e := range ces {
-		e := e
-		c, err := e.TipSetKey.Cid()
-		if err != nil {
-			return nil, fmt.Errorf("failed to get tipset cid: %w", err)
-		}
-
-		ev := &types.ActorEvent{
-			Entries:     e.Entries,
-			EmitterAddr: e.EmitterAddr,
-			Reverted:    e.Reverted,
-			Height:      e.Height,
-			TipSetCid:   c,
-			MsgCid:      e.MsgCid,
-		}
-
-		out = append(out, ev)
+		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, nil

node/impl/full/dummy.go

@@ -198,7 +198,7 @@ func (a *ActorEventDummy) GetActorEvents(ctx context.Context, filter *types.Acto
 	return nil, ErrActorEventModuleDisabled
 }
 
-func (a *ActorEventDummy) SubscribeActorEvents(ctx context.Context, filter *types.SubActorEventFilter) (<-chan *types.ActorEvent, error) {
+func (a *ActorEventDummy) SubscribeActorEvents(ctx context.Context, filter *types.ActorEventFilter) (<-chan *types.ActorEvent, error) {
 	return nil, ErrActorEventModuleDisabled
 }
 

node/impl/full/eth.go

@@ -1264,6 +1264,11 @@ func (e *EthEventHandler) EthGetFilterLogs(ctx context.Context, id ethtypes.EthF
 	return nil, xerrors.Errorf("wrong filter type")
 }
 
+// parseBlockRange is similar to actor event's parseHeightRange but with slightly different semantics
+//
+// * "block" instead of "height"
+// * strings that can have "latest" and "earliest" and nil
+// * hex strings for actual heights
 func parseBlockRange(heaviest abi.ChainEpoch, fromBlock, toBlock *string, maxRange abi.ChainEpoch) (minHeight abi.ChainEpoch, maxHeight abi.ChainEpoch, err error) {
 	if fromBlock == nil || *fromBlock == "latest" || len(*fromBlock) == 0 {
 		minHeight = heaviest

node/modules/actorevent.go

@@ -164,14 +164,14 @@ func EventFilterManager(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.Loc
 	}
 }
 
-func ActorEventHandler(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRepo, fx.Lifecycle, *filter.EventFilterManager, *store.ChainStore, *stmgr.StateManager, EventHelperAPI, *messagepool.MessagePool, full.StateAPI, full.ChainAPI) (*full.ActorEventHandler, error) {
+func ActorEventHandler(enable bool, fevmCfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRepo, fx.Lifecycle, *filter.EventFilterManager, *store.ChainStore, *stmgr.StateManager, EventHelperAPI, *messagepool.MessagePool, full.StateAPI, full.ChainAPI) (*full.ActorEventHandler, error) {
 	return func(mctx helpers.MetricsCtx, r repo.LockedRepo, lc fx.Lifecycle, fm *filter.EventFilterManager, cs *store.ChainStore, sm *stmgr.StateManager, evapi EventHelperAPI, mp *messagepool.MessagePool, stateapi full.StateAPI, chainapi full.ChainAPI) (*full.ActorEventHandler, error) {
 		ee := &full.ActorEventHandler{
-			MaxFilterHeightRange: abi.ChainEpoch(cfg.Events.MaxFilterHeightRange),
+			MaxFilterHeightRange: abi.ChainEpoch(fevmCfg.Events.MaxFilterHeightRange),
 			Chain:                cs,
 		}
 
-		if !cfg.EnableActorEventsAPI || cfg.Events.DisableRealTimeFilterAPI {
+		if !enable || fevmCfg.Events.DisableRealTimeFilterAPI {
 			// all Actor events functionality is disabled
 			return ee, nil
 		}