feat: record multisig approvals (#389)

* feat: record multisig approvals

Author: iand

chain/indexer.go

@@ -18,6 +18,8 @@ import (
 	"github.com/filecoin-project/sentinel-visor/metrics"
 	"github.com/filecoin-project/sentinel-visor/model"
 	visormodel "github.com/filecoin-project/sentinel-visor/model/visor"
+	"github.com/filecoin-project/sentinel-visor/tasks/messages"
+	"github.com/filecoin-project/sentinel-visor/tasks/msapprovals"
 )
 
 const (
@@ -31,6 +33,7 @@ const (
 	BlocksTask              = "blocks"              // task that extracts block data
 	MessagesTask            = "messages"            // task that extracts message data
 	ChainEconomicsTask      = "chaineconomics"      // task that extracts chain economics data
+	MultisigApprovalsTask   = "msapprovals"         // task that extracts multisig actor approvals
 )
 
 var log = logging.Logger("chain")
@@ -39,16 +42,17 @@ var _ TipSetObserver = (*TipSetIndexer)(nil)
 
 // A TipSetWatcher waits for tipsets and persists their block data into a database.
 type TipSetIndexer struct {
-	window          time.Duration
-	storage         model.Storage
-	processors      map[string]TipSetProcessor
-	actorProcessors map[string]ActorProcessor
-	name            string
-	persistSlot     chan struct{}
-	lastTipSet      *types.TipSet
-	node            lens.API
-	opener          lens.APIOpener
-	closer          lens.APICloser
+	window            time.Duration
+	storage           model.Storage
+	processors        map[string]TipSetProcessor
+	messageProcessors map[string]MessageProcessor
+	actorProcessors   map[string]ActorProcessor
+	name              string
+	persistSlot       chan struct{}
+	lastTipSet        *types.TipSet
+	node              lens.API
+	opener            lens.APIOpener
+	closer            lens.APICloser
 }
 
 // A TipSetIndexer extracts block, message and actor state data from a tipset and persists it to storage. Extraction
@@ -57,21 +61,22 @@ type TipSetIndexer struct {
 // indexer is used as the reporter in the visor_processing_reports table.
 func NewTipSetIndexer(o lens.APIOpener, d model.Storage, window time.Duration, name string, tasks []string) (*TipSetIndexer, error) {
 	tsi := &TipSetIndexer{
-		storage:         d,
-		window:          window,
-		name:            name,
-		persistSlot:     make(chan struct{}, 1), // allow one concurrent persistence job
-		processors:      map[string]TipSetProcessor{},
-		actorProcessors: map[string]ActorProcessor{},
-		opener:          o,
+		storage:           d,
+		window:            window,
+		name:              name,
+		persistSlot:       make(chan struct{}, 1), // allow one concurrent persistence job
+		processors:        map[string]TipSetProcessor{},
+		messageProcessors: map[string]MessageProcessor{},
+		actorProcessors:   map[string]ActorProcessor{},
+		opener:            o,
 	}
 
 	for _, task := range tasks {
 		switch task {
 		case BlocksTask:
 			tsi.processors[BlocksTask] = NewBlockProcessor()
 		case MessagesTask:
-			tsi.processors[MessagesTask] = NewMessageProcessor(o)
+			tsi.messageProcessors[MessagesTask] = messages.NewTask(o)
 		case ChainEconomicsTask:
 			tsi.processors[ChainEconomicsTask] = NewChainEconomicsProcessor(o)
 		case ActorStatesRawTask:
@@ -112,6 +117,8 @@ func NewTipSetIndexer(o lens.APIOpener, d model.Storage, window time.Duration, n
 				CodeV2: sa2builtin.MultisigActorCodeID,
 				CodeV3: sa3builtin.MultisigActorCodeID,
 			})
+		case MultisigApprovalsTask:
+			tsi.messageProcessors[MultisigApprovalsTask] = msapprovals.NewTask(o)
 		default:
 			return nil, xerrors.Errorf("unknown task: %s", task)
 		}
@@ -135,6 +142,8 @@ func (t *TipSetIndexer) TipSet(ctx context.Context, ts *types.TipSet) error {
 	}
 	defer cancel()
 
+	ll := log.With("height", int64(ts.Height()))
+
 	start := time.Now()
 
 	inFlight := 0
@@ -149,8 +158,8 @@ func (t *TipSetIndexer) TipSet(ctx context.Context, ts *types.TipSet) error {
 		go t.runProcessor(tctx, p, name, ts, results)
 	}
 
-	// Run each actor processing task concurrently if we have any and we've seen a previous tipset to compare with
-	if len(t.actorProcessors) > 0 {
+	// Run each actor or message processing task concurrently if we have any and we've seen a previous tipset to compare with
+	if len(t.actorProcessors) > 0 || len(t.messageProcessors) > 0 {
 
 		// Actor processors perform a diff between two tipsets so we need to keep track of parent and child
 		var parent, child *types.TipSet
@@ -181,36 +190,65 @@ func (t *TipSetIndexer) TipSet(ctx context.Context, ts *types.TipSet) error {
 				t.closer = closer
 			}
 
-			changes, err := t.node.StateChangedActors(tctx, parent.ParentState(), child.ParentState())
-			if err != nil {
-
-				terr := xerrors.Errorf("failed to extract actor changes: %w", err)
-				// We need to report that all actor tasks failed
-				for name := range t.actorProcessors {
-					report := &visormodel.ProcessingReport{
-						Height:         int64(ts.Height()),
-						StateRoot:      ts.ParentState().String(),
-						Reporter:       t.name,
-						Task:           name,
-						StartedAt:      start,
-						CompletedAt:    time.Now(),
-						Status:         visormodel.ProcessingStatusError,
-						ErrorsDetected: terr,
+			// If we have message processors then extract the messages and receipts
+			if len(t.messageProcessors) > 0 {
+				emsgs, err := t.node.GetExecutedMessagesForTipset(ctx, child, parent)
+				if err == nil {
+					// Start all the message processors
+					for name, p := range t.messageProcessors {
+						inFlight++
+						go t.runMessageProcessor(tctx, p, name, child, parent, emsgs, results)
+					}
+				} else {
+					ll.Errorw("failed to extract messages", "error", err)
+					terr := xerrors.Errorf("failed to extract messages: %w", err)
+					// We need to report that all message tasks failed
+					for name := range t.messageProcessors {
+						report := &visormodel.ProcessingReport{
+							Height:         int64(ts.Height()),
+							StateRoot:      ts.ParentState().String(),
+							Reporter:       t.name,
+							Task:           name,
+							StartedAt:      start,
+							CompletedAt:    time.Now(),
+							Status:         visormodel.ProcessingStatusError,
+							ErrorsDetected: terr,
+						}
+						taskOutputs[name] = model.PersistableList{report}
 					}
-					taskOutputs[name] = model.PersistableList{report}
 				}
-				return terr
 			}
 
-			for name, p := range t.actorProcessors {
-				inFlight++
-				go t.runActorProcessor(tctx, p, name, child, parent, changes, results)
+			// If we have actor processors then find actors that have changed state
+			if len(t.actorProcessors) > 0 {
+				changes, err := t.node.StateChangedActors(tctx, parent.ParentState(), child.ParentState())
+				if err == nil {
+					for name, p := range t.actorProcessors {
+						inFlight++
+						go t.runActorProcessor(tctx, p, name, child, parent, changes, results)
+					}
+				} else {
+					ll.Errorw("failed to extract actor changes", "error", err)
+					terr := xerrors.Errorf("failed to extract actor changes: %w", err)
+					// We need to report that all actor tasks failed
+					for name := range t.actorProcessors {
+						report := &visormodel.ProcessingReport{
+							Height:         int64(ts.Height()),
+							StateRoot:      ts.ParentState().String(),
+							Reporter:       t.name,
+							Task:           name,
+							StartedAt:      start,
+							CompletedAt:    time.Now(),
+							Status:         visormodel.ProcessingStatusError,
+							ErrorsDetected: terr,
+						}
+						taskOutputs[name] = model.PersistableList{report}
+					}
+				}
 			}
 		}
 	}
 
-	ll := log.With("height", int64(ts.Height()))
-
 	// Wait for all tasks to complete
 	for inFlight > 0 {
 		res := <-results
@@ -326,6 +364,28 @@ func (t *TipSetIndexer) runProcessor(ctx context.Context, p TipSetProcessor, nam
 	}
 }
 
+func (t *TipSetIndexer) runMessageProcessor(ctx context.Context, p MessageProcessor, name string, ts, pts *types.TipSet, emsgs []*lens.ExecutedMessage, results chan *TaskResult) {
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.TaskType, name))
+	stats.Record(ctx, metrics.TipsetHeight.M(int64(ts.Height())))
+	stop := metrics.Timer(ctx, metrics.ProcessingDuration)
+	defer stop()
+
+	data, report, err := p.ProcessMessages(ctx, ts, pts, emsgs)
+	if err != nil {
+		stats.Record(ctx, metrics.ProcessingFailure.M(1))
+		results <- &TaskResult{
+			Task:  name,
+			Error: err,
+		}
+		return
+	}
+	results <- &TaskResult{
+		Task:   name,
+		Report: report,
+		Data:   data,
+	}
+}
+
 func (t *TipSetIndexer) runActorProcessor(ctx context.Context, p ActorProcessor, name string, ts, pts *types.TipSet, actors map[string]types.Actor, results chan *TaskResult) {
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.TaskType, name))
 	stats.Record(ctx, metrics.TipsetHeight.M(int64(ts.Height())))
@@ -384,6 +444,14 @@ type TipSetProcessor interface {
 	Close() error
 }
 
+type MessageProcessor interface {
+	// ProcessMessages processes messages contained within a tipset. If error is non-nil then the processor encountered a fatal error.
+	// pts is the tipset containing the messages, ts is the tipset containing the receipts
+	// Any data returned must be accompanied by a processing report.
+	ProcessMessages(ctx context.Context, ts *types.TipSet, pts *types.TipSet, emsgs []*lens.ExecutedMessage) (model.Persistable, *visormodel.ProcessingReport, error)
+	Close() error
+}
+
 type ActorProcessor interface {
 	// ProcessActor processes a set of actors. If error is non-nil then the processor encountered a fatal error.
 	// Any data returned must be accompanied by a processing report.

lens/lotus/api.go

@@ -6,6 +6,7 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-bitfield"
 	"github.com/filecoin-project/lotus/api"
+	builtininit "github.com/filecoin-project/lotus/chain/actors/builtin/init"
 	miner "github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/state"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -220,6 +221,16 @@ func (aw *APIWrapper) GetExecutedMessagesForTipset(ctx context.Context, ts, pts
 		return nil, xerrors.Errorf("load parent state tree: %w", err)
 	}
 
+	initActor, err := stateTree.GetActor(builtininit.Address)
+	if err != nil {
+		return nil, xerrors.Errorf("getting init actor: %w", err)
+	}
+
+	initActorState, err := builtininit.Load(aw.Store(), initActor)
+	if err != nil {
+		return nil, xerrors.Errorf("loading init actor state: %w", err)
+	}
+
 	// Build a lookup of actor codes
 	actorCodes := map[address.Address]cid.Cid{}
 	if err := stateTree.ForEach(func(a address.Address, act *types.Actor) error {
@@ -230,7 +241,12 @@ func (aw *APIWrapper) GetExecutedMessagesForTipset(ctx context.Context, ts, pts
 	}
 
 	getActorCode := func(a address.Address) cid.Cid {
-		c, ok := actorCodes[a]
+		ra, found, err := initActorState.ResolveAddress(a)
+		if err != nil || !found {
+			return cid.Undef
+		}
+
+		c, ok := actorCodes[ra]
 		if ok {
 			return c
 		}

lens/util/repo.go

@@ -10,6 +10,7 @@ import (
 	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/api"
+	builtininit "github.com/filecoin-project/lotus/chain/actors/builtin/init"
 	"github.com/filecoin-project/lotus/chain/state"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/store"
@@ -24,14 +25,15 @@ import (
 	"github.com/filecoin-project/lotus/node/impl"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	"github.com/filecoin-project/lotus/node/repo"
-	"github.com/filecoin-project/sentinel-visor/lens"
 	"github.com/filecoin-project/specs-actors/actors/runtime/proof"
 	"github.com/filecoin-project/specs-actors/actors/util/adt"
 	"github.com/ipfs/go-cid"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	peer "github.com/libp2p/go-libp2p-core/peer"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/sentinel-visor/lens"
 )
 
 type APIOpener struct {
@@ -240,6 +242,16 @@ func GetExecutedMessagesForTipset(ctx context.Context, cs *store.ChainStore, ts,
 		return nil, xerrors.Errorf("load state tree: %w", err)
 	}
 
+	initActor, err := stateTree.GetActor(builtininit.Address)
+	if err != nil {
+		return nil, xerrors.Errorf("getting init actor: %w", err)
+	}
+
+	initActorState, err := builtininit.Load(cs.Store(ctx), initActor)
+	if err != nil {
+		return nil, xerrors.Errorf("loading init actor state: %w", err)
+	}
+
 	// Build a lookup of actor codes
 	actorCodes := map[address.Address]cid.Cid{}
 	if err := stateTree.ForEach(func(a address.Address, act *types.Actor) error {
@@ -250,7 +262,12 @@ func GetExecutedMessagesForTipset(ctx context.Context, cs *store.ChainStore, ts,
 	}
 
 	getActorCode := func(a address.Address) cid.Cid {
-		c, ok := actorCodes[a]
+		ra, found, err := initActorState.ResolveAddress(a)
+		if err != nil || !found {
+			return cid.Undef
+		}
+
+		c, ok := actorCodes[ra]
 		if ok {
 			return c
 		}

model/msapprovals/msapprovals.go

@@ -0,0 +1,48 @@
+package msapprovals
+
+import (
+	"context"
+
+	"go.opencensus.io/tag"
+
+	"github.com/filecoin-project/sentinel-visor/metrics"
+	"github.com/filecoin-project/sentinel-visor/model"
+)
+
+type MultisigApproval struct {
+	tableName      struct{} `pg:"multisig_approvals"` // nolint: structcheck,unused
+	Height         int64    `pg:",pk,notnull,use_zero"`
+	StateRoot      string   `pg:",pk,notnull"`
+	MultisigID     string   `pg:",pk,notnull"`
+	Message        string   `pg:",pk,notnull"`       // cid of message
+	Method         uint64   `pg:",notnull,use_zero"` // method number used for the approval 2=propose, 3=approve
+	Approver       string   `pg:",pk,notnull"`       // address of signer that triggerd approval
+	Threshold      uint64   `pg:",notnull,use_zero"`
+	InitialBalance string   `pg:"type:numeric,notnull"`
+	Signers        []string `pg:",notnull"`
+	GasUsed        int64    `pg:",use_zero"`
+	TransactionID  int64    `pg:",notnull,use_zero"`
+	To             string   `pg:",use_zero"`            // address funds will move to in transaction
+	Value          string   `pg:"type:numeric,notnull"` // amount of funds moved in transaction
+}
+
+func (ma *MultisigApproval) Persist(ctx context.Context, s model.StorageBatch) error {
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "multisig_approvals"))
+	stop := metrics.Timer(ctx, metrics.PersistDuration)
+	defer stop()
+
+	return s.PersistModel(ctx, ma)
+}
+
+type MultisigApprovalList []*MultisigApproval
+
+func (mal MultisigApprovalList) Persist(ctx context.Context, s model.StorageBatch) error {
+	if len(mal) == 0 {
+		return nil
+	}
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "multisig_approvals"))
+	stop := metrics.Timer(ctx, metrics.PersistDuration)
+	defer stop()
+
+	return s.PersistModel(ctx, mal)
+}