test(chain): refactor and test chain economics extraction

chain/economics.go

@@ -8,8 +8,8 @@ import (
 
 	"github.com/filecoin-project/sentinel-visor/lens"
 	"github.com/filecoin-project/sentinel-visor/model"
-	chainmodel "github.com/filecoin-project/sentinel-visor/model/chain"
 	visormodel "github.com/filecoin-project/sentinel-visor/model/visor"
+	"github.com/filecoin-project/sentinel-visor/tasks/chain"
 )
 
 type ChainEconomicsProcessor struct {
@@ -40,24 +40,15 @@ func (p *ChainEconomicsProcessor) ProcessTipSet(ctx context.Context, ts *types.T
 		StateRoot: ts.ParentState().String(),
 	}
 
-	supply, err := p.node.StateVMCirculatingSupplyInternal(ctx, ts.Key())
+	ce, err := chain.ExtractChainEconomicsModel(ctx, p.node, ts)
 	if err != nil {
-		log.Errorw("error received while fetching circulating supply messages, closing lens", "error", err)
+		log.Errorw("error received while extracting chain economics, closing lens", "error", err)
 		if cerr := p.Close(); cerr != nil {
 			log.Errorw("error received while closing lens", "error", cerr)
 		}
 		return nil, nil, err
 	}
 
-	ce := &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(),
-	}
-
 	return ce, report, nil
 }
 

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
@@ -134,27 +141,38 @@ func (p *ChainEconomics) processItem(ctx context.Context, node lens.API, item *v
 		return xerrors.Errorf("get tipset: %w", err)
 	}
 
-	supply, err := node.StateVMCirculatingSupplyInternal(ctx, tsk)
+	ce, err := ExtractChainEconomicsModel(ctx, node, ts)
 	if err != nil {
-		return err
-	}
-
-	ce := &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(),
+		return xerrors.Errorf("extracting chain economics model: %w", err)
 	}
 
-	log.Debugw("persisting tipset", "height", int64(ts.Height()))
+	log.Debugw("persisting tipset", "height", item.Height)
 
-	if err := p.storage.DB.RunInTransaction(ctx, func(tx *pg.Tx) error {
+	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
 }
+
+type ChainEconomicsLens interface {
+	StateVMCirculatingSupplyInternal(context.Context, types.TipSetKey) (api.CirculatingSupply, error)
+}
+
+func ExtractChainEconomicsModel(ctx context.Context, node ChainEconomicsLens, ts *types.TipSet) (*chainmodel.ChainEconomics, error) {
+	supply, err := node.StateVMCirculatingSupplyInternal(ctx, ts.Key())
+	if err != nil {
+		return nil, xerrors.Errorf("get circulating supply: %w", err)
+	}
+
+	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(),
+	}, nil
+}

tasks/chain/economics_test.go

@@ -0,0 +1,48 @@
+package chain
+
+import (
+	"context"
+	"testing"
+
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/filecoin-project/sentinel-visor/testutil"
+)
+
+type MockedChainEconomicsLens struct {
+	mock.Mock
+}
+
+func (m *MockedChainEconomicsLens) 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 TestEconomicsModelExtraction(t *testing.T) {
+	ctx := context.Background()
+
+	expectedTs := testutil.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(MockedChainEconomicsLens)
+	mockedLens.On("StateVMCirculatingSupplyInternal", ctx, expectedTs.Key()).Return(expectedCircSupply, nil)
+
+	em, err := ExtractChainEconomicsModel(ctx, mockedLens, expectedTs)
+	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)
+}

testutil/filtypes.go

@@ -0,0 +1,33 @@
+package testutil
+
+import (
+	"testing"
+
+	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/crypto"
+	"github.com/filecoin-project/lotus/chain/types"
+	tutils "github.com/filecoin-project/specs-actors/support/testing"
+	"github.com/ipfs/go-cid"
+)
+
+func FakeTipset(t testing.TB) *types.TipSet {
+	bh := FakeBlockHeader(t, 1, RandomCid())
+	ts, err := types.NewTipSet([]*types.BlockHeader{bh})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ts
+}
+
+func FakeBlockHeader(t testing.TB, height int64, stateRoot cid.Cid) *types.BlockHeader {
+	return &types.BlockHeader{
+		Miner:                 tutils.NewIDAddr(t, 123),
+		Height:                abi.ChainEpoch(height),
+		ParentStateRoot:       stateRoot,
+		Messages:              RandomCid(),
+		ParentMessageReceipts: RandomCid(),
+		BlockSig:              &crypto.Signature{Type: crypto.SigTypeBLS},
+		BLSAggregate:          &crypto.Signature{Type: crypto.SigTypeBLS},
+		Timestamp:             0,
+	}
+}