Make Snap sector picker deadline-aware (#173)

* snap: Avoid picking sectors in immutable deadlines

* snap: Use row iterator for candidates

* snap: Don't submit in immutable deadlines

* snap: Clone rows when not needed

Author: magik6k

harmony/harmonydb/sql/20240425-sector_meta.sql

@@ -23,6 +23,10 @@ CREATE TABLE sectors_meta (
     -- is_cc BOOLEAN NOT NULL DEFAULT (complex condition),
     -- expiration_epoch BIGINT, (null = not crawled)
 
+    -- Added in 20240826-sector-partition.sql
+    -- deadline BIGINT, (null = not crawled)
+    -- partition BIGINT, (null = not crawled)
+
     PRIMARY KEY (sp_id, sector_num)
 );
 

harmony/harmonydb/sql/20240611-snap-pipeline.sql

@@ -47,6 +47,10 @@ CREATE TABLE sectors_snap_pipeline (
     -- failed_reason varchar(20) not null default '',
     -- failed_reason_msg text not null default '',
 
+    -- added in 20240826-sector-partition.sql
+    -- when not null delays scheduling of the submit task
+    -- submit_after TIMESTAMP WITH TIME ZONE,
+
     FOREIGN KEY (sp_id, sector_number) REFERENCES sectors_meta (sp_id, sector_num),
     PRIMARY KEY (sp_id, sector_number)
 );

harmony/harmonydb/sql/20240826-sector-partition.sql

@@ -0,0 +1,11 @@
+ALTER TABLE sectors_meta ADD COLUMN deadline BIGINT;
+ALTER TABLE sectors_meta ADD COLUMN partition BIGINT;
+
+-- index on deadline/partition/spid/sectornum
+CREATE INDEX sectors_meta_deadline_partition_spid_sectornum_index ON sectors_meta(deadline, partition, sp_id, sector_num);
+
+-- schedule delay in case the sector got into an immutable deadline
+ALTER TABLE sectors_snap_pipeline ADD COLUMN submit_after TIMESTAMP WITH TIME ZONE;
+
+-- force sector metadata refresh
+DELETE FROM harmony_task_singletons WHERE task_name = 'SectorMetadata' and task_id IS NULL;

lib/curiochain/epoch.go

@@ -0,0 +1,17 @@
+package curiochain
+
+import (
+	"time"
+
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/lotus/build/buildconstants"
+	"github.com/filecoin-project/lotus/chain/types"
+)
+
+func EpochTime(curr *types.TipSet, e abi.ChainEpoch) time.Time {
+	diff := int64(buildconstants.BlockDelaySecs) * int64(curr.Height()-e)
+	curTs := curr.MinTimestamp() // unix seconds
+
+	return time.Unix(int64(curTs)+diff, 0)
+}

market/deal_ingest_seal.go

@@ -16,6 +16,7 @@ import (
 	"github.com/filecoin-project/go-padreader"
 	"github.com/filecoin-project/go-state-types/abi"
 	verifregtypes "github.com/filecoin-project/go-state-types/builtin/v9/verifreg"
+	"github.com/filecoin-project/go-state-types/dline"
 	"github.com/filecoin-project/go-state-types/network"
 
 	"github.com/filecoin-project/curio/harmony/harmonydb"
@@ -46,6 +47,10 @@ type PieceIngesterApi interface {
 	StateGetAllocationIdForPendingDeal(ctx context.Context, dealId abi.DealID, tsk types.TipSetKey) (verifregtypes.AllocationId, error)
 	StateLookupID(context.Context, address.Address, types.TipSetKey) (address.Address, error)
 	StateSectorGetInfo(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
+
+	StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok types.TipSetKey) (*miner.SectorLocation, error)
+	StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tsk types.TipSetKey) ([]api.Partition, error)
+	StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*dline.Info, error)
 }
 
 type openSector struct {

market/deal_ingest_snap.go

@@ -31,6 +31,11 @@ import (
 
 const IdealEndEpochBuffer = 2 * builtin.EpochsInDay
 
+// assuming that snap takes up to 20min to get to submitting the message we want to avoid sectors from deadlines which will
+// become immutable in the next 20min (40 epochs)
+// NOTE: Don't set this value to more than one deadline (60 epochs)
+var SnapImmutableDeadlineEpochsBuffer = abi.ChainEpoch(40)
+
 type PieceIngesterSnap struct {
 	ctx                 context.Context
 	db                  *harmonydb.DB
@@ -276,6 +281,25 @@ func (p *PieceIngesterSnap) AllocatePieceToSector(ctx context.Context, maddr add
 		}
 	}
 
+	// non-mutable deadline is the current deadline and the next one. Doesn't matter if the current one was proven or not.
+
+	curDeadline, err := p.api.StateMinerProvingDeadline(ctx, p.miner, types.EmptyTSK)
+	if err != nil {
+		return api.SectorOffset{}, xerrors.Errorf("getting proving deadline: %w", err)
+	}
+
+	dlIdxImmutableCur := curDeadline.Index
+	dlIdxImmutableNext := (curDeadline.Index + 1) % curDeadline.WPoStPeriodDeadlines
+
+	// The deadline which might become mutable soon
+	dlIdxImmutableNextNext := dlIdxImmutableNext
+	epochsInDeadline := curDeadline.CurrentEpoch - curDeadline.Open // how many into the deadline we are
+	epochsPerDeadline := curDeadline.WPoStProvingPeriod / abi.ChainEpoch(curDeadline.WPoStPeriodDeadlines)
+
+	if epochsInDeadline >= epochsPerDeadline-SnapImmutableDeadlineEpochsBuffer {
+		dlIdxImmutableNextNext = (dlIdxImmutableNext + 1) % curDeadline.WPoStPeriodDeadlines
+	}
+
 	// Allocation to open sector failed, create a new sector and add the piece to it
 
 	// TX
@@ -291,15 +315,15 @@ func (p *PieceIngesterSnap) AllocatePieceToSector(ctx context.Context, maddr add
 
 	var num *int64
 	_, err = p.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) {
-		var candidates []struct {
+		type CandidateSector struct {
 			Sector     int64 `db:"sector_num"`
 			Expiration int64 `db:"expiration_epoch"`
 		}
 
 		// maxExpiration = maybe(max sector expiration_epoch)
 		// minExpiration = piece.DealSchedule.EndEpoch
 		// ideal expiration = minExpiration + 2 days
-		err = tx.Select(&candidates, `
+		rows, err := tx.Query(`
 			SELECT sm.sector_num, sm.expiration_epoch
 			FROM sectors_meta sm
 			LEFT JOIN sectors_snap_pipeline ssp on sm.sp_id = ssp.sp_id and sm.sector_num = ssp.sector_number
@@ -309,28 +333,83 @@ func (p *PieceIngesterSnap) AllocatePieceToSector(ctx context.Context, maddr add
 			  AND sm.expiration_epoch IS NOT NULL
 			  AND sm.expiration_epoch > $1
 			  AND ($2 = 0 OR sm.expiration_epoch < $2)
+			  AND deadline IS NOT NULL AND deadline NOT IN ($5, $6, $7)
 			ORDER BY ABS(sm.expiration_epoch - ($1 + $3))
-			LIMIT 10
-		`, int64(piece.DealSchedule.EndEpoch), maxExpiration, IdealEndEpochBuffer, p.mid)
+		`, int64(piece.DealSchedule.EndEpoch), maxExpiration, IdealEndEpochBuffer, p.mid, dlIdxImmutableCur, dlIdxImmutableNext, dlIdxImmutableNextNext)
 		if err != nil {
 			return false, xerrors.Errorf("allocating sector numbers: %w", err)
 		}
+		defer rows.Close()
 
-		if len(candidates) == 0 {
+		deadlineCache := map[uint64][]api.Partition{}
+		var tried int
+		var bestCandidate *CandidateSector
+
+		for rows.Next() {
+			var candidate CandidateSector
+			err := rows.Scan(&candidate.Sector, &candidate.Expiration)
+			if err != nil {
+				return false, xerrors.Errorf("scanning row: %w", err)
+			}
+			tried++
+
+			sloc, err := p.api.StateSectorPartition(ctx, p.miner, abi.SectorNumber(candidate.Sector), types.EmptyTSK)
+			if err != nil {
+				return false, xerrors.Errorf("getting sector locations: %w", err)
+			}
+
+			if _, ok := deadlineCache[sloc.Deadline]; !ok {
+				dls, err := p.api.StateMinerPartitions(ctx, p.miner, sloc.Deadline, types.EmptyTSK)
+				if err != nil {
+					return false, xerrors.Errorf("getting partitions: %w", err)
+				}
+
+				deadlineCache[sloc.Deadline] = dls
+			}
+
+			dl := deadlineCache[sloc.Deadline]
+			if len(dl) <= int(sloc.Partition) {
+				return false, xerrors.Errorf("partition %d not found in deadline %d", sloc.Partition, sloc.Deadline)
+			}
+			part := dl[sloc.Partition]
+
+			active, err := part.ActiveSectors.IsSet(uint64(candidate.Sector))
+			if err != nil {
+				return false, xerrors.Errorf("checking active sectors: %w", err)
+			}
+			if !active {
+				live, err1 := part.LiveSectors.IsSet(uint64(candidate.Sector))
+				faulty, err2 := part.FaultySectors.IsSet(uint64(candidate.Sector))
+				recovering, err3 := part.RecoveringSectors.IsSet(uint64(candidate.Sector))
+				if err1 != nil || err2 != nil || err3 != nil {
+					return false, xerrors.Errorf("checking sector status: %w, %w, %w", err1, err2, err3)
+				}
+
+				log.Debugw("sector not active, skipping", "sector", candidate.Sector, "live", live, "faulty", faulty, "recovering", recovering)
+				continue
+			}
+
+			bestCandidate = &candidate
+			break
+		}
+
+		if err := rows.Err(); err != nil {
+			return false, xerrors.Errorf("iterating rows: %w", err)
+		}
+
+		rows.Close()
+
+		if bestCandidate == nil {
 			minEpoch := piece.DealSchedule.EndEpoch
 			maxEpoch := abi.ChainEpoch(maxExpiration)
 
 			minEpochDays := (minEpoch - head.Height()) / builtin.EpochsInDay
 			maxEpochDays := (maxEpoch - head.Height()) / builtin.EpochsInDay
 
-			return false, xerrors.Errorf("no suitable sectors found, minEpoch: %d, maxEpoch: %d, minExpirationDays: %d, maxExpirationDays: %d", minEpoch, maxEpoch, minEpochDays, maxEpochDays)
+			return false, xerrors.Errorf("no suitable sectors found, minEpoch: %d, maxEpoch: %d, minExpirationDays: %d, maxExpirationDays: %d (avoiding deadlines %d,%d,%d)", minEpoch, maxEpoch, minEpochDays, maxEpochDays, dlIdxImmutableCur, dlIdxImmutableNext, dlIdxImmutableNextNext)
 		}
 
-		// todo - nice to have:
-		//  * check sector liveness
-		//  * check deadline mutable
-
-		candidate := candidates[0] // this one works best
+		candidate := *bestCandidate
 
 		si, err := p.api.StateSectorGetInfo(ctx, p.miner, abi.SectorNumber(candidate.Sector), types.EmptyTSK)
 		if err != nil {
@@ -356,6 +435,7 @@ func (p *PieceIngesterSnap) AllocatePieceToSector(ctx context.Context, maddr add
 			"dealStartEpoch", piece.DealSchedule.StartEpoch,
 			"dealEndEpoch", piece.DealSchedule.EndEpoch,
 			"maxExpiration", maxExpiration,
+			"avoidingDeadlines", []int{int(dlIdxImmutableCur), int(dlIdxImmutableNext), int(dlIdxImmutableNextNext)},
 		)
 
 		_, err = tx.Exec(`SELECT insert_snap_ddo_piece($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,

market/lmrpc/lmrpc.go

@@ -581,7 +581,7 @@ func maybeApplyBackpressure(tx *harmonydb.Tx, cfg config.CurioIngestConfig, ssiz
 		}
 
 		if cfg.MaxQueueDealSector != 0 && waitDealSectors+sectors > cfg.MaxQueueDealSector {
-			log.Infow("backpressure", "reason", "too many wait deal sectors", "wait_deal_sectors", waitDealSectors, "max", cfg.MaxQueueDealSector)
+			log.Infow("backpressure", "reason", "too many wait deal sectors", "wait_deal_sectors", waitDealSectors, "parking-sectors", sectors, "parking-pieces", len(pieceSizes), "max", cfg.MaxQueueDealSector)
 			return true, nil
 		}
 

tasks/metadata/task_sector_expirations.go

@@ -24,6 +24,7 @@ const SectorMetadataRefreshInterval = 191 * time.Minute
 
 type SectorMetadataNodeAPI interface {
 	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
+	StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok types.TipSetKey) (*miner.SectorLocation, error)
 }
 
 type SectorMetadata struct {
@@ -49,22 +50,26 @@ func (s *SectorMetadata) Do(taskID harmonytask.TaskID, stillOwned func() bool) (
 		SectorNum uint64 `db:"sector_num"`
 
 		Expiration *uint64 `db:"expiration_epoch"`
+
+		Partition *uint64 `db:"partition"`
+		Deadline  *uint64 `db:"deadline"`
 	}
 
-	if err := s.db.Select(ctx, &sectors, "select sp_id, sector_num, expiration_epoch from sectors_meta ORDER BY sp_id, sector_num"); err != nil {
+	if err := s.db.Select(ctx, &sectors, "select sp_id, sector_num, expiration_epoch, partition, deadline from sectors_meta ORDER BY sp_id, sector_num"); err != nil {
 		return false, xerrors.Errorf("get sector list: %w", err)
 	}
 
 	astor := adt.WrapStore(ctx, cbor.NewCborStore(s.bstore))
 	minerStates := map[abi.ActorID]miner.State{}
 
 	for _, sector := range sectors {
+		maddr, err := address.NewIDAddress(sector.SpID)
+		if err != nil {
+			return false, xerrors.Errorf("creating miner address: %w", err)
+		}
+
 		mstate, ok := minerStates[abi.ActorID(sector.SpID)]
 		if !ok {
-			maddr, err := address.NewIDAddress(sector.SpID)
-			if err != nil {
-				return false, xerrors.Errorf("creating miner address: %w", err)
-			}
 
 			act, err := s.api.StateGetActor(ctx, maddr, types.EmptyTSK)
 			if err != nil {
@@ -93,6 +98,20 @@ func (s *SectorMetadata) Do(taskID harmonytask.TaskID, stillOwned func() bool) (
 				return false, xerrors.Errorf("updating sector expiration: %w", err)
 			}
 		}
+
+		if sector.Partition == nil || sector.Deadline == nil {
+			loc, err := s.api.StateSectorPartition(ctx, maddr, abi.SectorNumber(sector.SectorNum), types.EmptyTSK)
+			if err != nil {
+				return false, xerrors.Errorf("getting sector partition: %w", err)
+			}
+
+			if loc != nil {
+				_, err := s.db.Exec(ctx, "update sectors_meta set partition = $1, deadline = $2 where sp_id = $3 and sector_num = $4", loc.Partition, loc.Deadline, sector.SpID, sector.SectorNum)
+				if err != nil {
+					return false, xerrors.Errorf("updating sector partition: %w", err)
+				}
+			}
+		}
 	}
 
 	return true, nil