wdpost: Test repro broken challenge skipping

itests/kit/ensemble.go

@@ -697,6 +697,8 @@ func (n *Ensemble) Start() *Ensemble {
 		auth := http.Header(nil)
 
 		remote := stores.NewRemote(localStore, m.MinerNode, auth, 20, &stores.DefaultPartialFileHandler{})
+		store := m.options.workerStorageOpt(remote)
+
 		fh := &stores.FetchHandler{Local: localStore, PfHandler: &stores.DefaultPartialFileHandler{}}
 		m.FetchHandler = fh.ServeHTTP
 
@@ -706,7 +708,7 @@ func (n *Ensemble) Start() *Ensemble {
 			LocalWorker: sectorstorage.NewLocalWorker(sectorstorage.WorkerConfig{
 				TaskTypes: m.options.workerTasks,
 				NoSwap:    false,
-			}, remote, localStore, m.MinerNode, m.MinerNode, wsts),
+			}, store, localStore, m.MinerNode, m.MinerNode, wsts),
 			LocalStore: localStore,
 			Storage:    lr,
 		}

itests/kit/node_opts.go

@@ -4,6 +4,7 @@ import (
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -37,6 +38,7 @@ type nodeOpts struct {
 	minerNoLocalSealing  bool // use worker
 
 	workerTasks      []sealtasks.TaskType
+	workerStorageOpt func(stores.Store) stores.Store
 }
 
 // DefaultNodeOpts are the default options that will be applied to test nodes.
@@ -46,6 +48,7 @@ var DefaultNodeOpts = nodeOpts{
 	sectorSize: abi.SectorSize(2 << 10), // 2KiB.
 
 	workerTasks:      []sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize},
+	workerStorageOpt: func(store stores.Store) stores.Store { return store },
 }
 
 // OptBuilder is used to create an option after some other node is already
@@ -174,3 +177,10 @@ func WithTaskTypes(tt []sealtasks.TaskType) NodeOpt {
 		return nil
 	}
 }
+
+func WithWorkerStorage(transform func(stores.Store) stores.Store) NodeOpt {
+	return func(opts *nodeOpts) error {
+		opts.workerStorageOpt = transform
+		return nil
+	}
+}

itests/worker_test.go

@@ -2,6 +2,10 @@ package itests
 
 import (
 	"context"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
+	"golang.org/x/xerrors"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -158,3 +162,123 @@ func TestWindowPostWorker(t *testing.T) {
 	require.Equal(t, p.MinerPower, p.TotalPower)
 	require.Equal(t, p.MinerPower.RawBytePower, types.NewInt(uint64(ssz)*uint64(sectors-1)))
 }
+
+type badWorkerStorage struct {
+	stores.Store
+
+	badsector uint64
+}
+
+func (bs *badWorkerStorage) GenerateSingleVanillaProof(ctx context.Context, minerID abi.ActorID, si storiface.PostSectorChallenge, ppt abi.RegisteredPoStProof) ([]byte, error) {
+	if atomic.LoadUint64(&bs.badsector) == uint64(si.SectorNumber) {
+		return nil, xerrors.New("no proof for you")
+	}
+	return bs.Store.GenerateSingleVanillaProof(ctx, minerID, si, ppt)
+}
+
+func TestWindowPostWorkerSkipBadSector(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	_ = logging.SetLogLevel("storageminer", "INFO")
+
+	sectors := 2 * 48 * 2
+
+	var badStore *badWorkerStorage
+
+	client, miner, _, ens := kit.EnsembleWorker(t,
+		kit.PresealSectors(sectors), // 2 sectors per partition, 2 partitions in all 48 deadlines
+		kit.LatestActorsAt(-1),
+		kit.ThroughRPC(),
+		kit.WithTaskTypes([]sealtasks.TaskType{sealtasks.TTGenerateWindowPoSt}),
+		kit.WithWorkerStorage(func(store stores.Store) stores.Store {
+			badStore = &badWorkerStorage{
+				Store:     store,
+				badsector: 10000,
+			}
+			return badStore
+		}))
+
+	maddr, err := miner.ActorAddress(ctx)
+	require.NoError(t, err)
+
+	di, err := client.StateMinerProvingDeadline(ctx, maddr, types.EmptyTSK)
+	require.NoError(t, err)
+
+	bm := ens.InterconnectAll().BeginMining(2 * time.Millisecond)[0]
+
+	di = di.NextNotElapsed()
+
+	t.Log("Running one proving period")
+	waitUntil := di.Open + di.WPoStChallengeWindow*2 + storage.SubmitConfidence
+	client.WaitTillChain(ctx, kit.HeightAtLeast(waitUntil))
+
+	t.Log("Waiting for post message")
+	bm.Stop()
+
+	var lastPending []*types.SignedMessage
+	for i := 0; i < 500; i++ {
+		lastPending, err = client.MpoolPending(ctx, types.EmptyTSK)
+		require.NoError(t, err)
+
+		if len(lastPending) > 0 {
+			break
+		}
+		time.Sleep(40 * time.Millisecond)
+	}
+
+	require.Greater(t, len(lastPending), 0)
+
+	t.Log("post message landed")
+
+	bm.MineBlocks(ctx, 2*time.Millisecond)
+
+	waitUntil = di.Open + di.WPoStChallengeWindow*3
+	t.Logf("End for head.Height > %d", waitUntil)
+
+	ts := client.WaitTillChain(ctx, kit.HeightAtLeast(waitUntil))
+	t.Logf("Now head.Height = %d", ts.Height())
+
+	p, err := client.StateMinerPower(ctx, maddr, types.EmptyTSK)
+	require.NoError(t, err)
+
+	ssz, err := miner.ActorSectorSize(ctx, maddr)
+	require.NoError(t, err)
+
+	require.Equal(t, p.MinerPower, p.TotalPower)
+	require.Equal(t, p.MinerPower.RawBytePower, types.NewInt(uint64(ssz)*uint64(sectors)))
+
+	di, err = client.StateMinerProvingDeadline(ctx, maddr, types.EmptyTSK)
+	require.NoError(t, err)
+
+	// Remove one sector in the next deadline (so it's skipped)
+	{
+		parts, err := client.StateMinerPartitions(ctx, maddr, di.Index+1, types.EmptyTSK)
+		require.NoError(t, err)
+		require.Greater(t, len(parts), 0)
+
+		secs := parts[0].AllSectors
+		n, err := secs.Count()
+		require.NoError(t, err)
+		require.Equal(t, uint64(2), n)
+
+		// Drop the sector
+		sid, err := secs.First()
+		require.NoError(t, err)
+
+		t.Logf("Drop sector %d; dl %d part %d", sid, di.Index+1, 0)
+
+		atomic.StoreUint64(&badStore.badsector, sid)
+		require.NoError(t, err)
+	}
+
+	waitUntil = di.Close + di.WPoStChallengeWindow
+	ts = client.WaitTillChain(ctx, kit.HeightAtLeast(waitUntil))
+	t.Logf("Now head.Height = %d", ts.Height())
+
+	p, err = client.StateMinerPower(ctx, maddr, types.EmptyTSK)
+	require.NoError(t, err)
+
+	require.Equal(t, p.MinerPower, p.TotalPower)
+	require.Equal(t, p.MinerPower.RawBytePower, types.NewInt(uint64(ssz)*uint64(sectors-1)))
+}