Built-in actor events first draft

2024-03-22 07:00:28 +01:00
parent 5cb25affdd
commit 43ed348635
29 changed files with 810 additions and 115 deletions
@@ -36,6 +36,7 @@ type FullNodeAPI struct {
 	full.SyncAPI
 	full.RaftAPI
 	full.EthAPI
+	full.ActorEventsAPI

 	DS          dtypes.MetadataDS
 	NetworkName dtypes.NetworkName
@@ -0,0 +1,171 @@
+package full
+
+import (
+	"context"
+	"fmt"
+
+	"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/chain/events/filter"
+	"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.SubActorEventFilter) (<-chan *types.ActorEvent, error)
+}
+
+var (
+	_ ActorEventAPI = *new(api.FullNode)
+	_ ActorEventAPI = *new(api.Gateway)
+)
+
+type ActorEvent struct {
+	EventFilterManager   *filter.EventFilterManager
+	MaxFilterHeightRange abi.ChainEpoch
+}
+
+var _ ActorEventAPI = (*ActorEvent)(nil)
+
+type ActorEventsAPI struct {
+	fx.In
+	ActorEventAPI
+}
+
+func (a *ActorEvent) GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+	if a.EventFilterManager == nil {
+		return nil, api.ErrNotSupported
+	}
+
+	// Create a temporary filter
+	f, err := a.EventFilterManager.Install(ctx, filter.MinEpoch, filter.MaxEpoch, cid.Undef, filter.Addresses, filter.Fields, false)
+	if err != nil {
+		return nil, err
+	}
+
+	evs, err := getCollected(ctx, f)
+	_ = a.EventFilterManager.Remove(ctx, f.ID())
+	return evs, err
+}
+
+func (a *ActorEvent) SubscribeActorEvents(ctx context.Context, f *types.SubActorEventFilter) (<-chan *types.ActorEvent, error) {
+	if a.EventFilterManager == nil {
+		return nil, api.ErrNotSupported
+	}
+	fm, err := a.EventFilterManager.Install(ctx, f.Filter.MinEpoch, f.Filter.MaxEpoch, cid.Undef, f.Filter.Addresses, f.Filter.Fields, false)
+	if err != nil {
+		return nil, err
+	}
+
+	out := make(chan *types.ActorEvent, 25)
+
+	go func() {
+		defer func() {
+			// Tell the caller we're done
+			close(out)
+
+			// Unsubscribe.
+			fm.ClearSubChannel()
+			_ = 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
+			}
+
+			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
+				}
+			}
+		}
+
+		in := make(chan interface{}, 256)
+		fm.SetSubChannel(in)
+
+		for {
+			select {
+			case val, ok := <-in:
+				if !ok {
+					// Shutting down.
+					return
+				}
+
+				ce, ok := val.(*filter.CollectedEvent)
+				if !ok {
+					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,
+					TipSetKey:   c,
+					MsgCid:      ce.MsgCid,
+				}
+
+				select {
+				case out <- ev:
+				default:
+					log.Errorf("closing event subscription due to slow reader")
+					return
+				}
+				if len(out) > 5 {
+					log.Warnf("event subscription is slow, has %d buffered entries", len(out))
+				}
+
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return out, nil
+}
+
+func getCollected(ctx context.Context, f *filter.EventFilter) ([]*types.ActorEvent, error) {
+	ces := f.TakeCollectedEvents(ctx)
+
+	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,
+			TipSetKey:   c,
+			MsgCid:      e.MsgCid,
+		}
+
+		out = append(out, ev)
+	}
+
+	return out, nil
+}
@@ -11,6 +11,7 @@ import (
 	"github.com/filecoin-project/go-state-types/abi"

 	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/types/ethtypes"
 )

@@ -188,3 +189,17 @@ func (e *EthModuleDummy) EthTraceReplayBlockTransactions(ctx context.Context, bl

 var _ EthModuleAPI = &EthModuleDummy{}
 var _ EthEventAPI = &EthModuleDummy{}
+
+var ErrActorEventModuleDisabled = errors.New("module disabled, enable with Fevm.EnableActorEventsAPI")
+
+type ActorEventDummy struct{}
+
+func (a *ActorEventDummy) GetActorEvents(ctx context.Context, filter *types.ActorEventFilter) ([]*types.ActorEvent, error) {
+	return nil, ErrActorEventModuleDisabled
+}
+
+func (a *ActorEventDummy) SubscribeActorEvents(ctx context.Context, filter *types.SubActorEventFilter) (<-chan *types.ActorEvent, error) {
+	return nil, ErrActorEventModuleDisabled
+}
+
+var _ ActorEventAPI = &ActorEventDummy{}
@@ -12,6 +12,7 @@ import (
 	"time"

 	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multicodec"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
@@ -1353,7 +1354,20 @@ func (e *EthEvent) installEthFilterSpec(ctx context.Context, filterSpec *ethtype
 		return nil, err
 	}

-	return e.EventFilterManager.Install(ctx, minHeight, maxHeight, tipsetCid, addresses, keys)
+	return e.EventFilterManager.Install(ctx, minHeight, maxHeight, tipsetCid, addresses, keysToKeysWithCodec(keys), true)
+}
+
+func keysToKeysWithCodec(keys map[string][][]byte) map[string][]types.ActorEventBlock {
+	keysWithCodec := make(map[string][]types.ActorEventBlock)
+	for k, v := range keys {
+		for _, vv := range v {
+			keysWithCodec[k] = append(keysWithCodec[k], types.ActorEventBlock{
+				Codec: uint64(multicodec.Raw),
+				Value: vv,
+			})
+		}
+	}
+	return keysWithCodec
 }

 func (e *EthEvent) EthNewFilter(ctx context.Context, filterSpec *ethtypes.EthFilterSpec) (ethtypes.EthFilterID, error) {
@@ -1527,7 +1541,7 @@ func (e *EthEvent) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethty
 			}
 		}

-		f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, addresses, keys)
+		f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, addresses, keysToKeysWithCodec(keys), true)
 		if err != nil {
 			// clean up any previous filters added and stop the sub
 			_, _ = e.EthUnsubscribe(ctx, sub.id)