perf: minor optimization of market actor diffing

go.mod

@@ -18,10 +18,12 @@ require (
 	github.com/go-pg/pgext v0.1.4
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/ipfs/go-cid v0.0.7
+	github.com/ipfs/go-ipld-cbor v0.0.5-0.20200428170625-a0bd04d3cbdf
 	github.com/ipfs/go-log/v2 v2.1.2-0.20200626104915-0016c0b4b3e4
 	github.com/lib/pq v1.8.0
 	github.com/mattn/go-isatty v0.0.12 // indirect
 	github.com/multiformats/go-multiaddr v0.3.1
+	github.com/multiformats/go-multihash v0.0.14
 	github.com/raulk/clock v1.1.0
 	github.com/prometheus/client_golang v1.6.0
 	github.com/stretchr/objx v0.2.0 // indirect

go.sum

@@ -153,7 +153,6 @@ github.com/coreos/go-systemd v0.0.0-20181012123002-c6f51f82210d/go.mod h1:F5haX7
 github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd/v22 v22.0.0/go.mod h1:xO0FLkIi5MaZafQlIrOotqXZ90ih+1atmu1JpKERPPk=
 github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
-github.com/cpuguy83/go-md2man v1.0.10 h1:BSKMNlYxDvnunlTymqtgONjNnaRV1sTpcovwwjF22jk=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.0 h1:EoUDS0afbrsXAZ9YQ9jdu/mZ2sXgT1/2yyNng4PGlyM=
@@ -1277,7 +1276,6 @@ github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6L
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rs/cors v1.6.0 h1:G9tHG9lebljV9mfp9SNPDL36nCDxmo3zTlAf1YgvzmI=
 github.com/rs/cors v1.6.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
-github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -1384,7 +1382,6 @@ github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/
 github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
-github.com/urfave/cli v1.22.1 h1:+mkCCcOFKPnCmVYVcURKps1Xe+3zP90gSYGNfRkjoIY=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/urfave/cli/v2 v2.0.0/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2caLfQ=
 github.com/urfave/cli/v2 v2.2.0 h1:JTTnM6wKzdA0Jqodd966MVj4vWbbquZykeX1sKbe2C4=

tasks/actorstate/actor.go

@@ -7,7 +7,6 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 	"go.opentelemetry.io/otel/api/global"
 
-	"github.com/filecoin-project/sentinel-visor/lens"
 	"github.com/filecoin-project/sentinel-visor/metrics"
 	"github.com/filecoin-project/sentinel-visor/model"
 	commonmodel "github.com/filecoin-project/sentinel-visor/model/actors/common"
@@ -18,7 +17,7 @@ import (
 // ActorExtractor extracts common actor state
 type ActorExtractor struct{}
 
-func (ActorExtractor) Extract(ctx context.Context, a ActorInfo, node lens.API) (model.Persistable, error) {
+func (ActorExtractor) Extract(ctx context.Context, a ActorInfo, node ActorStateAPI) (model.Persistable, error) {
 	ctx, span := global.Tracer("").Start(ctx, "ActorExtractor")
 	defer span.End()
 

tasks/actorstate/actorstate.go

@@ -8,8 +8,10 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
+	"github.com/filecoin-project/specs-actors/actors/util/adt"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/raulk/clock"
@@ -27,8 +29,6 @@ import (
 
 var log = logging.Logger("actorstate")
 
-var timeNow = time.Now
-
 const batchInterval = 100 * time.Millisecond // time to wait between batches
 
 type ActorInfo struct {
@@ -39,9 +39,19 @@ type ActorInfo struct {
 	ParentTipSet    types.TipSetKey
 }
 
+// ActorStateAPI is the minimal subset of lens.API that is needed for actor state extraction
+type ActorStateAPI interface {
+	ChainHasObj(ctx context.Context, c cid.Cid) (bool, error)
+	ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, error)
+	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
+	StateMinerPower(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*api.MinerPower, error)
+	StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error)
+	Store() adt.Store
+}
+
 // An ActorStateExtractor extracts actor state into a persistable format
 type ActorStateExtractor interface {
-	Extract(ctx context.Context, a ActorInfo, node lens.API) (model.Persistable, error)
+	Extract(ctx context.Context, a ActorInfo, node ActorStateAPI) (model.Persistable, error)
 }
 
 // All supported actor state extractors

tasks/actorstate/actorstate_test.go

@@ -0,0 +1,237 @@
+package actorstate
+
+import (
+	"context"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/go-state-types/crypto"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
+	bstore "github.com/filecoin-project/lotus/lib/blockstore"
+	samarket "github.com/filecoin-project/specs-actors/actors/builtin/market"
+	"github.com/filecoin-project/specs-actors/actors/util/adt"
+	"github.com/ipfs/go-cid"
+	cbornode "github.com/ipfs/go-ipld-cbor"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/sentinel-visor/testutil"
+)
+
+func mockTipset(minerAddr address.Address, timestamp uint64) (*types.TipSet, error) {
+	return types.NewTipSet([]*types.BlockHeader{{
+		Miner:                 minerAddr,
+		Height:                5,
+		ParentStateRoot:       testutil.RandomCid(),
+		Messages:              testutil.RandomCid(),
+		ParentMessageReceipts: testutil.RandomCid(),
+		BlockSig:              &crypto.Signature{Type: crypto.SigTypeBLS},
+		BLSAggregate:          &crypto.Signature{Type: crypto.SigTypeBLS},
+		Timestamp:             timestamp,
+	}})
+}
+
+var _ ActorStateAPI = (*MockAPI)(nil)
+
+type MockAPI struct {
+	actors map[actorKey]*types.Actor
+	bs     bstore.Blockstore
+	store  adt.Store
+}
+
+func NewMockAPI() *MockAPI {
+	bs := bstore.NewTemporarySync()
+	return &MockAPI{
+		bs:     bs,
+		actors: make(map[actorKey]*types.Actor),
+		store:  adt.WrapStore(context.Background(), cbornode.NewCborStore(bs)),
+	}
+}
+
+type actorKey struct {
+	tsk  types.TipSetKey
+	addr address.Address
+}
+
+func (m *MockAPI) Store() adt.Store {
+	return m.store
+}
+
+func (m *MockAPI) ChainHasObj(ctx context.Context, c cid.Cid) (bool, error) {
+	return m.bs.Has(c)
+}
+
+func (m *MockAPI) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
+	blk, err := m.bs.Get(c)
+	if err != nil {
+		return nil, xerrors.Errorf("blockstore get: %w", err)
+	}
+
+	return blk.RawData(), nil
+}
+
+func (m *MockAPI) StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error) {
+	act, err := m.StateGetActor(ctx, actor, tsk)
+	if err != nil {
+		return nil, xerrors.Errorf("getting actor: %w", err)
+	}
+
+	var state interface{}
+	if err := m.store.Get(ctx, act.Head, &state); err != nil {
+		return nil, xerrors.Errorf("getting actor head: %w", err)
+	}
+
+	return &api.ActorState{
+		Balance: act.Balance,
+		State:   state,
+	}, nil
+}
+
+func (m *MockAPI) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+	key := actorKey{
+		tsk:  tsk,
+		addr: actor,
+	}
+	act, ok := m.actors[key]
+	if !ok {
+		return nil, xerrors.Errorf("actor not found addr:%s tsk=%s", actor, tsk)
+	}
+	return act, nil
+}
+
+func (m *MockAPI) StateMinerPower(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*api.MinerPower, error) {
+	panic("not implemented yet")
+}
+
+// ----------------- MockAPI Helpers ----------------------------
+
+func (m *MockAPI) setActor(tsk types.TipSetKey, addr address.Address, actor *types.Actor) {
+	key := actorKey{
+		tsk:  tsk,
+		addr: addr,
+	}
+	m.actors[key] = actor
+}
+
+func (m *MockAPI) createMarketState(ctx context.Context, deals map[abi.DealID]*samarket.DealState, props map[abi.DealID]*samarket.DealProposal, balances map[address.Address]balance) (cid.Cid, error) {
+	dealRootCid, err := m.createDealAMT(deals)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	propRootCid, err := m.createProposalAMT(props)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	balancesCids, err := m.createBalanceTable(balances)
+	if err != nil {
+		return cid.Undef, err
+	}
+	state, err := m.newEmptyMarketState()
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	state.States = dealRootCid
+	state.Proposals = propRootCid
+	state.EscrowTable = balancesCids[0]
+	state.LockedTable = balancesCids[1]
+
+	stateCid, err := m.store.Put(ctx, state)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	return stateCid, nil
+}
+
+func (m *MockAPI) newEmptyMarketState() (*samarket.State, error) {
+	emptyArrayCid, err := adt.MakeEmptyArray(m.store).Root()
+	if err != nil {
+		return nil, err
+	}
+	emptyMap, err := adt.MakeEmptyMap(m.store).Root()
+	if err != nil {
+		return nil, err
+	}
+	return samarket.ConstructState(emptyArrayCid, emptyMap, emptyMap), nil
+}
+
+func (m *MockAPI) createDealAMT(deals map[abi.DealID]*samarket.DealState) (cid.Cid, error) {
+	root := adt.MakeEmptyArray(m.store)
+	for dealID, dealState := range deals {
+		err := root.Set(uint64(dealID), dealState)
+		if err != nil {
+			return cid.Undef, err
+		}
+	}
+	rootCid, err := root.Root()
+	if err != nil {
+		return cid.Undef, err
+	}
+	return rootCid, nil
+}
+
+func (m *MockAPI) createProposalAMT(props map[abi.DealID]*samarket.DealProposal) (cid.Cid, error) {
+	root := adt.MakeEmptyArray(m.store)
+	for dealID, prop := range props {
+		err := root.Set(uint64(dealID), prop)
+		if err != nil {
+			return cid.Undef, err
+		}
+	}
+	rootCid, err := root.Root()
+	if err != nil {
+		return cid.Undef, err
+	}
+	return rootCid, nil
+}
+
+func (m *MockAPI) createBalanceTable(balances map[address.Address]balance) ([2]cid.Cid, error) {
+	escrowMapRoot := adt.MakeEmptyMap(m.store)
+	escrowMapRootCid, err := escrowMapRoot.Root()
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+	escrowRoot, err := adt.AsBalanceTable(m.store, escrowMapRootCid)
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+
+	lockedMapRoot := adt.MakeEmptyMap(m.store)
+	lockedMapRootCid, err := lockedMapRoot.Root()
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+
+	lockedRoot, err := adt.AsBalanceTable(m.store, lockedMapRootCid)
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+
+	for addr, balance := range balances {
+		err := escrowRoot.Add(addr, big.Add(balance.available, balance.locked))
+		if err != nil {
+			return [2]cid.Cid{}, err
+		}
+
+		err = lockedRoot.Add(addr, balance.locked)
+		if err != nil {
+			return [2]cid.Cid{}, err
+		}
+
+	}
+	escrowRootCid, err := escrowRoot.Root()
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+
+	lockedRootCid, err := lockedRoot.Root()
+	if err != nil {
+		return [2]cid.Cid{}, err
+	}
+
+	return [2]cid.Cid{escrowRootCid, lockedRootCid}, nil
+}

tasks/actorstate/genesis_test.go

@@ -122,7 +122,6 @@ func TestGenesisProcessor(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotEqual(t, 0, count)
 	})
-
 }
 
 // truncateGenesisTables ensures the indexing tables are empty

tasks/actorstate/init.go

@@ -9,7 +9,6 @@ import (
 	"github.com/filecoin-project/lotus/chain/events/state"
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 
-	"github.com/filecoin-project/sentinel-visor/lens"
 	"github.com/filecoin-project/sentinel-visor/metrics"
 	"github.com/filecoin-project/sentinel-visor/model"
 	initmodel "github.com/filecoin-project/sentinel-visor/model/actors/init"
@@ -24,7 +23,7 @@ func init() {
 	Register(builtin.InitActorCodeID, InitExtractor{})
 }
 
-func (InitExtractor) Extract(ctx context.Context, a ActorInfo, node lens.API) (model.Persistable, error) {
+func (InitExtractor) Extract(ctx context.Context, a ActorInfo, node ActorStateAPI) (model.Persistable, error) {
 	ctx, span := global.Tracer("").Start(ctx, "InitExtractor")
 	defer span.End()