feat: add new task miner sector deal v2 (#1293)

* Add new task: miner_sector_deal_v2

chain/actors/builtin/market/actor.go.template

@@ -75,6 +75,8 @@ type State interface {
 
 	DealProposalsAmtBitwidth() int
 	DealStatesAmtBitwidth() int
+
+	GetProviderSectors() (map[abi.SectorID][]abi.DealID, error)
 }
 
 type BalanceTable interface {

chain/actors/builtin/market/state.go.template

@@ -306,4 +306,44 @@ func (s *state{{.v}}) Code() cid.Cid {
         }
         return markettypes.NewLabelFromBytes(bs)
     }
-{{end}}
+{{end}}
+
+func (s *state{{.v}}) GetProviderSectors() (map[abi.SectorID][]abi.DealID, error) {
+	{{if (le .v 12)}}
+	return nil, nil
+	{{else}}
+	sectorDeals, err := adt{{.v}}.AsMap(s.store, s.State.ProviderSectors, market{{.v}}.ProviderSectorsHamtBitwidth)
+	if err != nil {
+		return nil, err
+	}
+	var sectorMapRoot cbg.CborCid
+	providerSectors := make(map[abi.SectorID][]abi.DealID)
+	err = sectorDeals.ForEach(&sectorMapRoot, func(providerID string) error {
+		provider, err := abi.ParseUIntKey(providerID)
+		if err != nil {
+			return nil
+		}
+
+		sectorMap, err := adt{{.v}}.AsMap(s.store, cid.Cid(sectorMapRoot), market{{.v}}.ProviderSectorsHamtBitwidth)
+		if err != nil {
+			return err
+		}
+
+		var dealIDs market{{.v}}.SectorDealIDs
+		err = sectorMap.ForEach(&dealIDs, func(sectorID string) error {
+			sectorNumber, err := abi.ParseUIntKey(sectorID)
+			if err != nil {
+				return err
+			}
+
+			dealIDsCopy := make([]abi.DealID, len(dealIDs))
+			copy(dealIDsCopy, dealIDs)
+
+			providerSectors[abi.SectorID{Miner: abi.ActorID(provider), Number: abi.SectorNumber(sectorNumber)}] = dealIDsCopy
+			return nil
+		})
+		return err
+	})
+	return providerSectors, err
+	{{end}}
+   }

chain/indexer/integrated/processor/state.go

@@ -695,6 +695,11 @@ func MakeProcessors(api tasks.DataSource, indexerTasks []string) (*IndexerProces
 				marketactors.AllCodes(),
 				markettask.DealProposalExtractor{},
 			))
+		case tasktype.MinerSectorDealV2:
+			out.ActorProcessors[t] = actorstate.NewTask(api, actorstate.NewTypedActorExtractorMap(
+				marketactors.AllCodes(),
+				markettask.SectorDealStateExtractor{},
+			))
 
 			//
 			// Multisig

chain/indexer/integrated/processor/state_internal_test.go

@@ -50,7 +50,7 @@ func TestNewProcessor(t *testing.T) {
 	proc, err := New(nil, t.Name(), tasktype.AllTableTasks)
 	require.NoError(t, err)
 	require.Equal(t, t.Name(), proc.name)
-	require.Len(t, proc.actorProcessors, 25)
+	require.Len(t, proc.actorProcessors, 26)
 	require.Len(t, proc.tipsetProcessors, 10)
 	require.Len(t, proc.tipsetsProcessors, 15)
 	require.Len(t, proc.builtinProcessors, 1)

chain/indexer/integrated/processor/state_test.go

@@ -439,7 +439,7 @@ func TestMakeProcessorsAllTasks(t *testing.T) {
 	// If this test fails it indicates a new processor and/or task name was added and test should be created for it in one of the above test cases.
 	proc, err := processor.MakeProcessors(nil, append(tasktype.AllTableTasks, processor.BuiltinTaskName))
 	require.NoError(t, err)
-	require.Len(t, proc.ActorProcessors, 25)
+	require.Len(t, proc.ActorProcessors, 26)
 	require.Len(t, proc.TipsetProcessors, 10)
 	require.Len(t, proc.TipsetsProcessors, 15)
 	require.Len(t, proc.ReportProcessors, 1)

chain/indexer/tasktype/table_tasks.go

@@ -8,6 +8,7 @@ const (
 	DataCapBalance                 = "data_cap_balance"
 	MinerBeneficiary               = "miner_beneficiary"
 	MinerSectorDeal                = "miner_sector_deal"
+	MinerSectorDealV2              = "miner_sector_deal_v2"
 	MinerSectorInfoV7              = "miner_sector_infos_v7"
 	MinerSectorInfoV1_6            = "miner_sector_infos"
 	MinerSectorPost                = "miner_sector_post"
@@ -109,6 +110,7 @@ var AllTableTasks = []string{
 	FEVMActorDump,
 	MinerActorDump,
 	BuiltInActorEvent,
+	MinerSectorDealV2,
 }
 
 var TableLookup = map[string]struct{}{
@@ -164,6 +166,7 @@ var TableLookup = map[string]struct{}{
 	FEVMActorDump:                  {},
 	MinerActorDump:                 {},
 	BuiltInActorEvent:              {},
+	MinerSectorDealV2:              {},
 }
 
 var TableComment = map[string]string{
@@ -219,6 +222,7 @@ var TableComment = map[string]string{
 	FEVMActorDump:                  ``,
 	MinerActorDump:                 ``,
 	BuiltInActorEvent:              ``,
+	MinerSectorDealV2:              ``,
 }
 
 var TableFieldComments = map[string]map[string]string{
@@ -447,4 +451,5 @@ var TableFieldComments = map[string]map[string]string{
 		"TotalLockedFunds": "Locked Funds",
 	},
 	BuiltInActorEvent: {},
+	MinerSectorDealV2: {},
 }

chain/indexer/tasktype/tasks.go

@@ -53,6 +53,7 @@ var TaskLookup = map[string][]string{
 	ActorStatesMarketTask: {
 		MarketDealProposal,
 		MarketDealState,
+		MinerSectorDealV2,
 	},
 	ActorStatesMultisigTask: {
 		MultisigTransaction,

chain/indexer/tasktype/tasks_test.go

@@ -39,7 +39,7 @@ func TestMakeTaskNamesAlias(t *testing.T) {
 		},
 		{
 			taskAlias: tasktype.ActorStatesMarketTask,
-			tasks:     []string{tasktype.MarketDealProposal, tasktype.MarketDealState},
+			tasks:     []string{tasktype.MarketDealProposal, tasktype.MarketDealState, tasktype.MinerSectorDealV2},
 		},
 		{
 			taskAlias: tasktype.ActorStatesMultisigTask,
@@ -102,7 +102,7 @@ func TestMakeAllTaskAliasNames(t *testing.T) {
 }
 
 func TestMakeAllTaskNames(t *testing.T) {
-	const TotalTableTasks = 52
+	const TotalTableTasks = 53
 	actual, err := tasktype.MakeTaskNames(tasktype.AllTableTasks)
 	require.NoError(t, err)
 	// if this test fails it means a new task name was added, update the above test

model/actors/miner/sectordealsV2.go

@@ -0,0 +1,45 @@
+package miner
+
+import (
+	"context"
+
+	"go.opencensus.io/tag"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/filecoin-project/lily/metrics"
+	"github.com/filecoin-project/lily/model"
+)
+
+type MinerSectorDealV2 struct {
+	tableName struct{} `pg:"miner_sector_deals_v2"` // nolint: structcheck
+
+	Height   int64  `pg:",pk,notnull,use_zero"`
+	MinerID  string `pg:",pk,notnull"`
+	SectorID uint64 `pg:",pk,use_zero"`
+	DealID   uint64 `pg:",pk,use_zero"`
+}
+
+func (ds *MinerSectorDealV2) Persist(ctx context.Context, s model.StorageBatch, _ model.Version) error {
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "miner_sector_deals_v2"))
+	metrics.RecordCount(ctx, metrics.PersistModel, 1)
+	return s.PersistModel(ctx, ds)
+}
+
+type MinerSectorDealListV2 []*MinerSectorDealV2
+
+func (ml MinerSectorDealListV2) Persist(ctx context.Context, s model.StorageBatch, _ model.Version) error {
+	ctx, span := otel.Tracer("").Start(ctx, "MinerSectorDealListV2.Persist")
+	if span.IsRecording() {
+		span.SetAttributes(attribute.Int("count", len(ml)))
+	}
+	defer span.End()
+
+	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Table, "miner_sector_deals_v2"))
+
+	if len(ml) == 0 {
+		return nil
+	}
+	metrics.RecordCount(ctx, metrics.PersistModel, len(ml))
+	return s.PersistModel(ctx, ml)
+}

schemas/v1/38_miner_sector_info_v2.go

@@ -0,0 +1,19 @@
+package v1
+
+func init() {
+	patches.Register(
+		38,
+		`
+	CREATE TABLE {{ .SchemaName | default "public"}}.miner_sector_deals_v2 (
+		miner_id text NOT NULL,
+		sector_id bigint NOT NULL,
+		deal_id bigint NOT NULL,
+		height bigint NOT NULL
+	);
+	ALTER TABLE ONLY {{ .SchemaName | default "public"}}.miner_sector_deals_v2 ADD CONSTRAINT miner_sector_deals_v2_pkey PRIMARY KEY (height, miner_id, sector_id, deal_id);
+	CREATE INDEX IF NOT EXISTS miner_sector_deals_height_idx ON {{ .SchemaName | default "public"}}.miner_sector_deals_v2 USING btree (height DESC);
+	CREATE INDEX IF NOT EXISTS miner_sector_deals_miner_id_idx ON {{ .SchemaName | default "public"}}.miner_sector_deals_v2 USING hash (miner_id);
+	CREATE INDEX IF NOT EXISTS miner_sector_deals_sector_id_idx ON {{ .SchemaName | default "public"}}.miner_sector_deals_v2 USING hash (sector_id);
+`,
+	)
+}

storage/sql.go

@@ -58,6 +58,7 @@ var Models = []interface{}{
 	(*miner.MinerFeeDebt)(nil),
 	(*miner.MinerLockedFund)(nil),
 	(*miner.MinerInfo)(nil),
+	(*miner.MinerSectorDealV2)(nil),
 
 	(*market.MarketDealProposal)(nil),
 	(*market.MarketDealState)(nil),