use pre-commit policy to set pre-commit sector expiry epoch

go.mod

@@ -5,6 +5,7 @@ go 1.13
 require (
 	github.com/filecoin-project/go-address v0.0.2-0.20200218010043-eb9bb40ed5be
 	github.com/filecoin-project/go-cbor-util v0.0.0-20191219014500-08c40a1e63a2
+	github.com/filecoin-project/go-fil-commcid v0.0.0-20200208005934-2b8bd03caca5
 	github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6
 	github.com/filecoin-project/go-paramfetch v0.0.2-0.20200218225740-47c639bab663 // indirect
 	github.com/filecoin-project/go-statemachine v0.0.0-20200226041606-2074af6d51d9

precommit_policy.go

@@ -0,0 +1,76 @@
+package sealing
+
+import (
+	"context"
+
+	"github.com/filecoin-project/specs-actors/actors/abi"
+)
+
+type PreCommitPolicy interface {
+	Expiration(ctx context.Context, pdis ...PieceWithOptionalDealInfo) (abi.ChainEpoch, error)
+}
+
+type Chain interface {
+	ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error)
+}
+
+// BasicPreCommitPolicy satisfies PreCommitPolicy. It has two modes:
+//
+// Mode 1: The sector contains a non-zero quantity of pieces with deal info
+// Mode 2: The sector contains no pieces with deal info
+//
+// The BasicPreCommitPolicy#Expiration method is given a slice of the pieces
+// which the miner has encoded into the sector, and from that slice picks either
+// the first or second mode.
+//
+// If we're in Mode 1: The pre-commit expiration epoch will be the maximum
+// deal end epoch of a piece in the sector.
+//
+// If we're in Mode 2: The pre-commit expiration epoch will be set to the
+// current epoch + the provided default duration.
+type BasicPreCommitPolicy struct {
+	api Chain
+
+	duration abi.ChainEpoch
+}
+
+// NewBasicPreCommitPolicy produces a BasicPreCommitPolicy
+func NewBasicPreCommitPolicy(api Chain, duration abi.ChainEpoch) BasicPreCommitPolicy {
+	return BasicPreCommitPolicy{
+		api:      api,
+		duration: duration,
+	}
+}
+
+// Expiration produces the pre-commit sector expiration epoch for an encoded
+// replica containing the provided enumeration of pieces and deals.
+func (p *BasicPreCommitPolicy) Expiration(ctx context.Context, pdis ...PieceWithOptionalDealInfo) (abi.ChainEpoch, error) {
+	_, epoch, err := p.api.ChainHead(ctx)
+	if err != nil {
+		return 0, nil
+	}
+
+	var end *abi.ChainEpoch
+
+	for _, p := range pdis {
+		if p.DealInfo == nil {
+			continue
+		}
+
+		if p.DealInfo.DealSchedule.EndEpoch < epoch {
+			log.Warnf("piece schedule %+v ended before current epoch %d", p, epoch)
+			continue
+		}
+
+		if end == nil || *end < p.DealInfo.DealSchedule.EndEpoch {
+			end = &p.DealInfo.DealSchedule.EndEpoch
+		}
+	}
+
+	if end == nil {
+		tmp := epoch + p.duration
+		end = &tmp
+	}
+
+	return *end, nil
+}

precommit_policy_test.go

@@ -0,0 +1,134 @@
+package sealing_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	commcid "github.com/filecoin-project/go-fil-commcid"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+
+	sealing "github.com/filecoin-project/storage-fsm"
+)
+
+type fakeChain struct {
+	h abi.ChainEpoch
+}
+
+func (f *fakeChain) ChainHead(ctx context.Context) (sealing.TipSetToken, abi.ChainEpoch, error) {
+	return []byte{1, 2, 3}, f.h, nil
+}
+
+func TestBasicPolicyEmptySector(t *testing.T) {
+	policy := sealing.NewBasicPreCommitPolicy(&fakeChain{
+		h: abi.ChainEpoch(55),
+	}, 10)
+
+	exp, err := policy.Expiration(context.Background())
+	require.NoError(t, err)
+
+	assert.Equal(t, 65, int(exp))
+}
+
+func TestBasicPolicyMostConstrictiveSchedule(t *testing.T) {
+	policy := sealing.NewBasicPreCommitPolicy(&fakeChain{
+		h: abi.ChainEpoch(55),
+	}, 100)
+
+	pieces := []sealing.PieceWithOptionalDealInfo{
+		{
+			Piece: abi.PieceInfo{
+				Size:     abi.PaddedPieceSize(1024),
+				PieceCID: commcid.ReplicaCommitmentV1ToCID([]byte{1, 2, 3}),
+			},
+			DealInfo: &sealing.DealInfo{
+				DealID: abi.DealID(42),
+				DealSchedule: sealing.DealSchedule{
+					StartEpoch: abi.ChainEpoch(70),
+					EndEpoch:   abi.ChainEpoch(75),
+				},
+			},
+		},
+		{
+			Piece: abi.PieceInfo{
+				Size:     abi.PaddedPieceSize(1024),
+				PieceCID: commcid.ReplicaCommitmentV1ToCID([]byte{1, 2, 3}),
+			},
+			DealInfo: &sealing.DealInfo{
+				DealID: abi.DealID(43),
+				DealSchedule: sealing.DealSchedule{
+					StartEpoch: abi.ChainEpoch(80),
+					EndEpoch:   abi.ChainEpoch(100),
+				},
+			},
+		},
+	}
+
+	exp, err := policy.Expiration(context.Background(), pieces...)
+	require.NoError(t, err)
+
+	assert.Equal(t, 100, int(exp))
+}
+
+func TestBasicPolicyIgnoresExistingScheduleIfExpired(t *testing.T) {
+	policy := sealing.NewBasicPreCommitPolicy(&fakeChain{
+		h: abi.ChainEpoch(55),
+	}, 100)
+
+	pieces := []sealing.PieceWithOptionalDealInfo{
+		{
+			Piece: abi.PieceInfo{
+				Size:     abi.PaddedPieceSize(1024),
+				PieceCID: commcid.ReplicaCommitmentV1ToCID([]byte{1, 2, 3}),
+			},
+			DealInfo: &sealing.DealInfo{
+				DealID: abi.DealID(44),
+				DealSchedule: sealing.DealSchedule{
+					StartEpoch: abi.ChainEpoch(1),
+					EndEpoch:   abi.ChainEpoch(10),
+				},
+			},
+		},
+	}
+
+	exp, err := policy.Expiration(context.Background(), pieces...)
+	require.NoError(t, err)
+
+	assert.Equal(t, 155, int(exp))
+}
+
+func TestMissingDealIsIgnored(t *testing.T) {
+	policy := sealing.NewBasicPreCommitPolicy(&fakeChain{
+		h: abi.ChainEpoch(55),
+	}, 100)
+
+	pieces := []sealing.PieceWithOptionalDealInfo{
+		{
+			Piece: abi.PieceInfo{
+				Size:     abi.PaddedPieceSize(1024),
+				PieceCID: commcid.ReplicaCommitmentV1ToCID([]byte{1, 2, 3}),
+			},
+			DealInfo: &sealing.DealInfo{
+				DealID: abi.DealID(44),
+				DealSchedule: sealing.DealSchedule{
+					StartEpoch: abi.ChainEpoch(1),
+					EndEpoch:   abi.ChainEpoch(10),
+				},
+			},
+		},
+		{
+			Piece: abi.PieceInfo{
+				Size:     abi.PaddedPieceSize(1024),
+				PieceCID: commcid.ReplicaCommitmentV1ToCID([]byte{1, 2, 3}),
+			},
+			DealInfo: nil,
+		},
+	}
+
+	exp, err := policy.Expiration(context.Background(), pieces...)
+	require.NoError(t, err)
+
+	assert.Equal(t, 155, int(exp))
+}

sealing.go

@@ -49,9 +49,11 @@ type Sealing struct {
 	sc      SectorIDCounter
 	verif   ffiwrapper.Verifier
 	tktFn   TicketFn
+
+	pcp PreCommitPolicy
 }
 
-func New(api SealingAPI, events Events, maddr address.Address, worker address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, tktFn TicketFn) *Sealing {
+func New(api SealingAPI, events Events, maddr address.Address, worker address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, tktFn TicketFn, pcp PreCommitPolicy) *Sealing {
 	s := &Sealing{
 		api:    api,
 		events: events,
@@ -62,6 +64,7 @@ func New(api SealingAPI, events Events, maddr address.Address, worker address.Ad
 		sc:     sc,
 		verif:  verif,
 		tktFn:  tktFn,
+		pcp:    pcp,
 	}
 
 	s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{})

states.go

@@ -112,8 +112,13 @@ func (m *Sealing) handlePreCommitting(ctx statemachine.Context, sector SectorInf
 		}
 	}
 
+	expiration, err := m.pcp.Expiration(ctx.Context(), sector.PiecesWithOptionalDealInfo...)
+	if err != nil {
+		return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("handlePreCommitting: failed to compute pre-commit expiry: %w", err)})
+	}
+
 	params := &miner.SectorPreCommitInfo{
-		Expiration:      10000000, // TODO: implement expiration
+		Expiration:      expiration,
 		SectorNumber:    sector.SectorNumber,
 		RegisteredProof: sector.SectorType,