test(chain): test chain economics processor

go.sum

@@ -162,6 +162,7 @@ github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f/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=
@@ -714,6 +715,7 @@ github.com/ipld/go-ipld-prime-proto v0.1.0 h1:j7gjqrfwbT4+gXpHwEx5iMssma3mnctC7Y
 github.com/ipld/go-ipld-prime-proto v0.1.0/go.mod h1:11zp8f3sHVgIqtb/c9Kr5ZGqpnCLF1IVTNOez9TopzE=
 github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52 h1:QG4CGBqCeuBo6aZlGAamSkxWdgWfZGeE49eUOWJPA4c=
 github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52/go.mod h1:fdg+/X9Gg4AsAIzWpEHwnqd+QY3b7lajxyjE1m4hkq4=
+github.com/jackc/chunkreader v1.0.0 h1:4s39bBR8ByfqH+DKm8rQA3E1LHZWB9XWcrz8fqaZbe0=
 github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo=
 github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
@@ -732,6 +734,7 @@ github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2 h1:JVX6jT/XfzNqIjye47
 github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2/go.mod h1:fGZlG77KXmcq05nJLRkk0+p82V8B8Dw8KN2/V9c/OAE=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgproto3 v1.1.0 h1:FYYE4yRw+AgI8wXIinMlNjBbp/UitDJwfj5LqqewP1A=
 github.com/jackc/pgproto3 v1.1.0/go.mod h1:eR5FA3leWg7p9aeAqi37XOTgTIbkABlvcPB3E5rlc78=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190420180111-c116219b62db/go.mod h1:bhq50y+xrl9n5mRYyCBFKkpRVTLYJVWeCc+mEAI3yXA=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190609003834-432c2951c711/go.mod h1:uH0AWtUmuShn0bcesswc4aBTWGvw0cAxIJp+6OB//Wg=
@@ -1430,6 +1433,7 @@ github.com/rs/cors v1.6.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/rs/zerolog v1.13.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU=
 github.com/rs/zerolog v1.15.0/go.mod h1:xYTKnLHcpfU2225ny5qZjxnj9NvkumZYjJHlAThCjNc=
+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/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
@@ -1539,6 +1543,7 @@ 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/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2caLfQ=

storage/sql.go

@@ -707,3 +707,7 @@ func (d *Database) Persist(ctx context.Context, p model.PersistableWithTx) error
 		return p.PersistWithTx(ctx, tx)
 	})
 }
+
+func (d *Database) RunInTransaction(ctx context.Context, fn func(tx *pg.Tx) error) error {
+	return d.DB.RunInTransaction(ctx, fn)
+}

tasks/chain/economics.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/go-pg/pg/v10"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/raulk/clock"
@@ -16,7 +18,6 @@ import (
 	"github.com/filecoin-project/sentinel-visor/metrics"
 	chainmodel "github.com/filecoin-project/sentinel-visor/model/chain"
 	"github.com/filecoin-project/sentinel-visor/model/visor"
-	"github.com/filecoin-project/sentinel-visor/storage"
 	"github.com/filecoin-project/sentinel-visor/wait"
 )
 
@@ -27,7 +28,13 @@ const (
 
 var log = logging.Logger("chain")
 
-func NewChainEconomicsProcessor(d *storage.Database, opener lens.APIOpener, leaseLength time.Duration, batchSize int, minHeight, maxHeight int64) *ChainEconomics {
+type EconomicsStorage interface {
+	RunInTransaction(ctx context.Context, fn func(tx *pg.Tx) error) error
+	MarkTipSetEconomicsComplete(ctx context.Context, tipset string, height int64, completedAt time.Time, errorsDetected string) error
+	LeaseTipSetEconomics(ctx context.Context, claimUntil time.Time, batchSize int, minHeight, maxHeight int64) (visor.ProcessingTipSetList, error)
+}
+
+func NewChainEconomicsProcessor(d EconomicsStorage, opener lens.APIOpener, leaseLength time.Duration, batchSize int, minHeight, maxHeight int64) *ChainEconomics {
 	return &ChainEconomics{
 		opener:      opener,
 		storage:     d,
@@ -43,7 +50,7 @@ func NewChainEconomicsProcessor(d *storage.Database, opener lens.APIOpener, leas
 // persists the results to the database.
 type ChainEconomics struct {
 	opener      lens.APIOpener
-	storage     *storage.Database
+	storage     EconomicsStorage
 	leaseLength time.Duration // length of time to lease work for
 	batchSize   int           // number of tipsets to lease in a batch
 	minHeight   int64         // limit processing to tipsets equal to or above this height
@@ -116,7 +123,12 @@ func (p *ChainEconomics) processBatch(ctx context.Context, node lens.API) (bool,
 	return false, nil
 }
 
-func (p *ChainEconomics) processItem(ctx context.Context, node lens.API, item *visor.ProcessingTipSet) error {
+type EconomicsProcessItemsLens interface {
+	ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
+	StateVMCirculatingSupplyInternal(context.Context, types.TipSetKey) (api.CirculatingSupply, error)
+}
+
+func (p *ChainEconomics) processItem(ctx context.Context, node EconomicsProcessItemsLens, item *visor.ProcessingTipSet) error {
 	ctx, span := global.Tracer("").Start(ctx, "ChainEconomics.processItem")
 	defer span.End()
 	span.SetAttributes(label.Any("height", item.Height), label.Any("tipset", item.TipSet))
@@ -129,32 +141,39 @@ func (p *ChainEconomics) processItem(ctx context.Context, node lens.API, item *v
 		return xerrors.Errorf("get tipsetkey: %w", err)
 	}
 
+	ce, err := extractChainEconomicsModel(ctx, node, tsk)
+	if err != nil {
+		return xerrors.Errorf("extracting chain economics model: %w", err)
+	}
+
+	log.Debugw("persisting tipset", "height", item.Height)
+
+	if err := p.storage.RunInTransaction(ctx, func(tx *pg.Tx) error {
+		return ce.PersistWithTx(ctx, tx)
+	}); err != nil {
+		return xerrors.Errorf("persist: %w", err)
+	}
+
+	return nil
+}
+
+func extractChainEconomicsModel(ctx context.Context, node EconomicsProcessItemsLens, tsk types.TipSetKey) (*chainmodel.ChainEconomics, error) {
 	ts, err := node.ChainGetTipSet(ctx, tsk)
 	if err != nil {
-		return xerrors.Errorf("get tipset: %w", err)
+		return nil, xerrors.Errorf("get tipset: %w", err)
 	}
 
 	supply, err := node.StateVMCirculatingSupplyInternal(ctx, tsk)
 	if err != nil {
-		return err
+		return nil, xerrors.Errorf("get circulating supply: %w", err)
 	}
 
-	ce := &chainmodel.ChainEconomics{
+	return &chainmodel.ChainEconomics{
 		ParentStateRoot: ts.ParentState().String(),
 		VestedFil:       supply.FilVested.String(),
 		MinedFil:        supply.FilMined.String(),
 		BurntFil:        supply.FilBurnt.String(),
 		LockedFil:       supply.FilLocked.String(),
 		CirculatingFil:  supply.FilCirculating.String(),
-	}
-
-	log.Debugw("persisting tipset", "height", int64(ts.Height()))
-
-	if err := p.storage.DB.RunInTransaction(ctx, func(tx *pg.Tx) error {
-		return ce.PersistWithTx(ctx, tx)
-	}); err != nil {
-		return xerrors.Errorf("persist: %w", err)
-	}
-
-	return nil
+	}, nil
 }

tasks/chain/economics_test.go

@@ -0,0 +1,74 @@
+package chain
+
+import (
+	"context"
+	"testing"
+
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/crypto"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
+	tutils "github.com/filecoin-project/specs-actors/support/testing"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/filecoin-project/sentinel-visor/testutil"
+)
+
+type MockedEconomicsLens struct {
+	mock.Mock
+}
+
+func (m *MockedEconomicsLens) ChainGetTipSet(ctx context.Context, key types.TipSetKey) (*types.TipSet, error) {
+	args := m.Called(ctx, key)
+	return args.Get(0).(*types.TipSet), args.Error(1)
+}
+
+func (m *MockedEconomicsLens) StateVMCirculatingSupplyInternal(ctx context.Context, key types.TipSetKey) (api.CirculatingSupply, error) {
+	args := m.Called(ctx, key)
+	return args.Get(0).(api.CirculatingSupply), args.Error(1)
+}
+
+func fakeTipset(t testing.TB) *types.TipSet {
+	bh := &types.BlockHeader{
+		Miner:                 tutils.NewIDAddr(t, 123),
+		Height:                1,
+		ParentStateRoot:       testutil.RandomCid(),
+		Messages:              testutil.RandomCid(),
+		ParentMessageReceipts: testutil.RandomCid(),
+		BlockSig:              &crypto.Signature{Type: crypto.SigTypeBLS},
+		BLSAggregate:          &crypto.Signature{Type: crypto.SigTypeBLS},
+		Timestamp:             0,
+	}
+	ts, err := types.NewTipSet([]*types.BlockHeader{bh})
+	if err != nil {
+		panic(err)
+	}
+	return ts
+}
+
+func TestEconomicsModelExtraction(t *testing.T) {
+	ctx := context.Background()
+
+	expectedTs := fakeTipset(t)
+	expectedCircSupply := api.CirculatingSupply{
+		FilVested:      abi.NewTokenAmount(1),
+		FilMined:       abi.NewTokenAmount(2),
+		FilBurnt:       abi.NewTokenAmount(3),
+		FilLocked:      abi.NewTokenAmount(4),
+		FilCirculating: abi.NewTokenAmount(5),
+	}
+
+	mockedLens := new(MockedEconomicsLens)
+	mockedLens.On("ChainGetTipSet", ctx, expectedTs.Key()).Return(expectedTs, nil)
+	mockedLens.On("StateVMCirculatingSupplyInternal", ctx, expectedTs.Key()).Return(expectedCircSupply, nil)
+
+	em, err := extractChainEconomicsModel(ctx, mockedLens, expectedTs.Key())
+	assert.NoError(t, err)
+	assert.EqualValues(t, expectedTs.ParentState().String(), em.ParentStateRoot)
+	assert.EqualValues(t, expectedCircSupply.FilBurnt.String(), em.BurntFil)
+	assert.EqualValues(t, expectedCircSupply.FilMined.String(), em.MinedFil)
+	assert.EqualValues(t, expectedCircSupply.FilVested.String(), em.VestedFil)
+	assert.EqualValues(t, expectedCircSupply.FilLocked.String(), em.LockedFil)
+	assert.EqualValues(t, expectedCircSupply.FilCirculating.String(), em.CirculatingFil)
+}