Merge pull request #1045 from filecoin-project/feat/mock-sectorbuilder

WIP: implement mock sectorbuilder for easier storage miner testing

cmd/lotus-bench/main.go

@@ -188,7 +188,7 @@ func main() {
 
 				log.Info("Running replication...")
 				pieces := []sectorbuilder.PublicPieceInfo{pi}
-				pco, err := sb.SealPreCommit(i, ticket, pieces)
+				pco, err := sb.SealPreCommit(context.TODO(), i, ticket, pieces)
 				if err != nil {
 					return xerrors.Errorf("commit: %w", err)
 				}
@@ -206,7 +206,7 @@ func main() {
 				}
 
 				log.Info("Generating PoRep for sector")
-				proof, err := sb.SealCommit(i, ticket, seed, pieces, pco)
+				proof, err := sb.SealCommit(context.TODO(), i, ticket, seed, pieces, pco)
 				if err != nil {
 					return err
 				}

cmd/lotus-seal-worker/sub.go

@@ -103,7 +103,7 @@ func (w *worker) processTask(ctx context.Context, task sectorbuilder.WorkerTask)
 
 	switch task.Type {
 	case sectorbuilder.WorkerPreCommit:
-		rspco, err := w.sb.SealPreCommit(task.SectorID, task.SealTicket, task.Pieces)
+		rspco, err := w.sb.SealPreCommit(ctx, task.SectorID, task.SealTicket, task.Pieces)
 		if err != nil {
 			return errRes(xerrors.Errorf("precomitting: %w", err))
 		}
@@ -117,7 +117,7 @@ func (w *worker) processTask(ctx context.Context, task sectorbuilder.WorkerTask)
 			return errRes(xerrors.Errorf("pushing precommited data: %w", err))
 		}
 	case sectorbuilder.WorkerCommit:
-		proof, err := w.sb.SealCommit(task.SectorID, task.SealTicket, task.SealSeed, task.Pieces, task.Rspco)
+		proof, err := w.sb.SealCommit(ctx, task.SectorID, task.SealTicket, task.SealSeed, task.Pieces, task.Rspco)
 		if err != nil {
 			return errRes(xerrors.Errorf("comitting: %w", err))
 		}

cmd/lotus-seed/seed/seed.go

@@ -69,7 +69,7 @@ func PreSeal(maddr address.Address, ssize uint64, offset uint64, sectors int, sb
 
 		fmt.Printf("sector-id: %d, piece info: %v", sid, pi)
 
-		pco, err := sb.SealPreCommit(sid, ticket, []sectorbuilder.PublicPieceInfo{pi})
+		pco, err := sb.SealPreCommit(context.TODO(), sid, ticket, []sectorbuilder.PublicPieceInfo{pi})
 		if err != nil {
 			return nil, xerrors.Errorf("commit: %w", err)
 		}

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03
 	github.com/filecoin-project/go-data-transfer v0.0.0-20191219005021-4accf56bd2ce
 	github.com/filecoin-project/go-paramfetch v0.0.1
-	github.com/filecoin-project/go-sectorbuilder v0.0.0-20200107220006-3361d30ea5ab
+	github.com/filecoin-project/go-sectorbuilder v0.0.0-20200109194458-9656ce473254
 	github.com/filecoin-project/go-statestore v0.0.0-20200102200712-1f63c701c1e5
 	github.com/gbrlsnchs/jwt/v3 v3.0.0-beta.1
 	github.com/go-ole/go-ole v1.2.4 // indirect
@@ -47,7 +47,7 @@ require (
 	github.com/ipfs/go-ipfs-routing v0.1.0
 	github.com/ipfs/go-ipld-cbor v0.0.3
 	github.com/ipfs/go-ipld-format v0.0.2
-	github.com/ipfs/go-log v1.0.1 // indirect
+	github.com/ipfs/go-log v1.0.1
 	github.com/ipfs/go-log/v2 v2.0.2
 	github.com/ipfs/go-merkledag v0.2.4
 	github.com/ipfs/go-path v0.0.7

go.sum

@@ -118,8 +118,8 @@ github.com/filecoin-project/go-paramfetch v0.0.0-20200102181131-b20d579f2878/go.
 github.com/filecoin-project/go-paramfetch v0.0.0-20200102181131-b20d579f2878/go.mod h1:40kI2Gv16mwcRsHptI3OAV4nlOEU7wVDc4RgMylNFjU=
 github.com/filecoin-project/go-paramfetch v0.0.1 h1:gV7bs5YaqlgpGFMiLxInGK2L1FyCXUE0rimz4L7ghoE=
 github.com/filecoin-project/go-paramfetch v0.0.1/go.mod h1:fZzmf4tftbwf9S37XRifoJlz7nCjRdIrMGLR07dKLCc=
-github.com/filecoin-project/go-sectorbuilder v0.0.0-20200107220006-3361d30ea5ab h1:bsrBNO1LwnhOLxPEXlSPal/WuY61mLJUCHYyD0NayHg=
+github.com/filecoin-project/go-sectorbuilder v0.0.0-20200109194458-9656ce473254 h1:4IvlPad82JaNBtqh8fEAUIKWv8I3tguAJjGvUyHNZS4=
-github.com/filecoin-project/go-sectorbuilder v0.0.0-20200107220006-3361d30ea5ab/go.mod h1:3OZ4E3B2OuwhJjtxR4r7hPU9bCfB+A+hm4alLEsaeDc=
+github.com/filecoin-project/go-sectorbuilder v0.0.0-20200109194458-9656ce473254/go.mod h1:3OZ4E3B2OuwhJjtxR4r7hPU9bCfB+A+hm4alLEsaeDc=
 github.com/filecoin-project/go-statestore v0.0.0-20200102200712-1f63c701c1e5 h1:NZXq90YlfakSmB2/84dGr0AVmKYFA97+yyViBIgTFbk=
 github.com/filecoin-project/go-statestore v0.0.0-20200102200712-1f63c701c1e5/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
@@ -283,12 +283,12 @@ github.com/ipfs/go-log v0.0.1/go.mod h1:kL1d2/hzSpI0thNYjiKfjanbVNU+IIGA/WnNESY9
 github.com/ipfs/go-log v1.0.0 h1:BW3LQIiZzpNyolt84yvKNCd3FU+AK4VDw1hnHR+1aiI=
 github.com/ipfs/go-log v1.0.0/go.mod h1:JO7RzlMK6rA+CIxFMLOuB6Wf5b81GDiKElL7UPSIKjA=
 github.com/ipfs/go-log v1.0.1 h1:5lIEEOQTk/vd1WuPFBRqz2mcp+5G1fMVcW+Ib/H5Hfo=
-github.com/ipfs/go-log v1.0.1/go.mod h1:HuWlQttfN6FWNHRhlY5yMk/lW7evQC0HHGOxEwMRR8I=
-github.com/ipfs/go-log/v2 v2.0.1 h1:mnR9XFltezAtO8A6tj5U7nKkRzhEQNEw/wT11U2HhPM=
-github.com/ipfs/go-log/v2 v2.0.1/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
 github.com/ipfs/go-log v1.0.1 h1:5lIEEOQTk/vd1WuPFBRqz2mcp+5G1fMVcW+Ib/H5Hfo=
 github.com/ipfs/go-log v1.0.1/go.mod h1:HuWlQttfN6FWNHRhlY5yMk/lW7evQC0HHGOxEwMRR8I=
+github.com/ipfs/go-log v1.0.1/go.mod h1:HuWlQttfN6FWNHRhlY5yMk/lW7evQC0HHGOxEwMRR8I=
 github.com/ipfs/go-log/v2 v2.0.1 h1:mnR9XFltezAtO8A6tj5U7nKkRzhEQNEw/wT11U2HhPM=
+github.com/ipfs/go-log/v2 v2.0.1 h1:mnR9XFltezAtO8A6tj5U7nKkRzhEQNEw/wT11U2HhPM=
+github.com/ipfs/go-log/v2 v2.0.1/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
 github.com/ipfs/go-log/v2 v2.0.1/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
 github.com/ipfs/go-log/v2 v2.0.2 h1:xguurydRdfKMJjKyxNXNU8lYP0VZH1NUwJRwUorjuEw=
 github.com/ipfs/go-log/v2 v2.0.2/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
@@ -879,9 +879,9 @@ golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191125144606-a911d9008d1f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191216173652-a0e659d51361 h1:RIIXAeV6GvDBuADKumTODatUqANFZ+5BPMnzsy4hulY=
 golang.org/x/tools v0.0.0-20191216173652-a0e659d51361/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
-golang.org/x/tools v0.0.0-20191125144606-a911d9008d1f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200108195415-316d2f248479 h1:csuS+MHeEA2eWhyjQCMaPMq4z1+/PohkBSjJZHSIbOE=
 golang.org/x/tools v0.0.0-20200108195415-316d2f248479/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

node/builder.go

@@ -238,7 +238,7 @@ func Online() Option {
 
 		// Storage miner
 		ApplyIf(func(s *Settings) bool { return s.nodeType == repo.StorageMiner },
-			Override(new(*sectorbuilder.SectorBuilder), modules.SectorBuilder),
+			Override(new(storage.SectorBuilder), modules.SectorBuilder),
 			Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
 			Override(new(storage.TicketFn), modules.SealTicketGen),
 			Override(new(*storage.Miner), modules.StorageMiner),

node/impl/storminer.go

@@ -3,12 +3,13 @@ package impl
 import (
 	"context"
 	"encoding/json"
-	"github.com/filecoin-project/lotus/api/apistruct"
 	"io"
 	"mime"
 	"net/http"
 	"os"
 
+	"github.com/filecoin-project/lotus/api/apistruct"
+
 	"github.com/gorilla/mux"
 	files "github.com/ipfs/go-ipfs-files"
 
@@ -25,7 +26,7 @@ type StorageMinerAPI struct {
 	CommonAPI
 
 	SectorBuilderConfig *sectorbuilder.Config
-	SectorBuilder       *sectorbuilder.SectorBuilder
+	SectorBuilder       storage.SectorBuilder
 	SectorBlocks        *sectorblocks.SectorBlocks
 
 	Miner      *storage.Miner

node/modules/storageminer.go

@@ -95,7 +95,7 @@ func SectorBuilderConfig(storagePath string, threads uint, noprecommit, nocommit
 	}
 }
 
-func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sb *sectorbuilder.SectorBuilder, tktFn storage.TicketFn) (*storage.Miner, error) {
+func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sb storage.SectorBuilder, tktFn storage.TicketFn) (*storage.Miner, error) {
 	maddr, err := minerAddrFromDS(ds)
 	if err != nil {
 		return nil, err

node/node_test.go

@@ -236,7 +236,7 @@ func builder(t *testing.T, nFull int, storage []int) ([]test.TestNode, []test.Te
 			t.Fatal(err)
 		}
 
-		if err := sma.SectorBuilder.ImportFrom(osb, false); err != nil {
+		if err := sma.SectorBuilder.(*sectorbuilder.SectorBuilder).ImportFrom(osb, false); err != nil {
 			t.Fatal(err)
 		}
 

storage/fpost_sched.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"sync"
 
-	sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
 	"go.opencensus.io/trace"
 	"golang.org/x/xerrors"
 
@@ -20,7 +19,7 @@ const StartConfidence = 4 // TODO: config
 
 type fpostScheduler struct {
 	api storageMinerApi
-	sb  *sectorbuilder.SectorBuilder
+	sb  SectorBuilder
 
 	actor  address.Address
 	worker address.Address

storage/miner.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"context"
 	"errors"
+	"io"
 	"time"
 
 	"github.com/ipfs/go-cid"
@@ -36,7 +37,7 @@ type Miner struct {
 	worker address.Address
 
 	// Sealing
-	sb      *sectorbuilder.SectorBuilder
+	sb      SectorBuilder
 	sectors *statestore.StateStore
 	tktFn   TicketFn
 
@@ -71,7 +72,24 @@ type storageMinerApi interface {
 	WalletHas(context.Context, address.Address) (bool, error)
 }
 
-func NewMiner(api storageMinerApi, addr address.Address, h host.Host, ds datastore.Batching, sb *sectorbuilder.SectorBuilder, tktFn TicketFn) (*Miner, error) {
+type SectorBuilder interface {
+	RateLimit() func()
+	AddPiece(uint64, uint64, io.Reader, []uint64) (sectorbuilder.PublicPieceInfo, error)
+	SectorSize() uint64
+	AcquireSectorId() (uint64, error)
+	Scrub(sectorbuilder.SortedPublicSectorInfo) []*sectorbuilder.Fault
+	GenerateFallbackPoSt(sectorbuilder.SortedPublicSectorInfo, [sectorbuilder.CommLen]byte, []uint64) ([]sectorbuilder.EPostCandidate, []byte, error)
+	SealPreCommit(context.Context, uint64, sectorbuilder.SealTicket, []sectorbuilder.PublicPieceInfo) (sectorbuilder.RawSealPreCommitOutput, error)
+	SealCommit(context.Context, uint64, sectorbuilder.SealTicket, sectorbuilder.SealSeed, []sectorbuilder.PublicPieceInfo, sectorbuilder.RawSealPreCommitOutput) ([]byte, error)
+
+	// Not so sure about these being on the interface
+	GetPath(string, string) (string, error)
+	WorkerStats() sectorbuilder.WorkerStats
+	AddWorker(context.Context, sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error)
+	TaskDone(context.Context, uint64, sectorbuilder.SealRes) error
+}
+
+func NewMiner(api storageMinerApi, addr address.Address, h host.Host, ds datastore.Batching, sb SectorBuilder, tktFn TicketFn) (*Miner, error) {
 	return &Miner{
 		api: api,
 
@@ -147,8 +165,8 @@ type SectorBuilderEpp struct {
 	sb *sectorbuilder.SectorBuilder
 }
 
-func NewElectionPoStProver(sb *sectorbuilder.SectorBuilder) *SectorBuilderEpp {
+func NewElectionPoStProver(sb SectorBuilder) *SectorBuilderEpp {
-	return &SectorBuilderEpp{sb}
+	return &SectorBuilderEpp{sb.(*sectorbuilder.SectorBuilder)}
 }
 
 var _ gen.ElectionPoStProver = (*SectorBuilderEpp)(nil)

storage/sbmock/sbmock.go

@@ -0,0 +1,254 @@
+package sbmock
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"sync"
+
+	"github.com/filecoin-project/go-sectorbuilder"
+	"golang.org/x/xerrors"
+)
+
+func randComm() [sectorbuilder.CommLen]byte {
+	var out [sectorbuilder.CommLen]byte
+	rand.Read(out[:])
+	return out
+}
+
+type SBMock struct {
+	sectors      map[uint64]*sectorState
+	sectorSize   uint64
+	nextSectorID uint64
+	rateLimit    chan struct{}
+
+	lk sync.Mutex
+}
+
+func NewMockSectorBuilder(threads int, ssize uint64) *SBMock {
+	return &SBMock{
+		sectors:      make(map[uint64]*sectorState),
+		sectorSize:   ssize,
+		nextSectorID: 0,
+		rateLimit:    make(chan struct{}, threads),
+	}
+}
+
+const (
+	statePacking = iota
+	statePreCommit
+	stateCommit
+)
+
+type sectorState struct {
+	pieces [][]byte
+	failed bool
+
+	state int
+
+	lk sync.Mutex
+}
+
+func (sb *SBMock) RateLimit() func() {
+	sb.rateLimit <- struct{}{}
+
+	// TODO: probably want to copy over rate limit code
+	return func() {
+		<-sb.rateLimit
+	}
+}
+
+func (sb *SBMock) AddPiece(size uint64, sectorId uint64, r io.Reader, existingPieces []uint64) (sectorbuilder.PublicPieceInfo, error) {
+	sb.lk.Lock()
+	ss, ok := sb.sectors[sectorId]
+	if !ok {
+		ss = &sectorState{
+			state: statePacking,
+		}
+		sb.sectors[sectorId] = ss
+	}
+	sb.lk.Unlock()
+	ss.lk.Lock()
+	defer ss.lk.Unlock()
+
+	b, err := ioutil.ReadAll(r)
+	if err != nil {
+		return sectorbuilder.PublicPieceInfo{}, err
+	}
+
+	ss.pieces = append(ss.pieces, b)
+	return sectorbuilder.PublicPieceInfo{
+		Size: size,
+		// TODO: should we compute a commP? maybe do it when we need it
+	}, nil
+}
+
+func (sb *SBMock) SectorSize() uint64 {
+	return sb.sectorSize
+}
+
+func (sb *SBMock) AcquireSectorId() (uint64, error) {
+	sb.lk.Lock()
+	defer sb.lk.Unlock()
+	id := sb.nextSectorID
+	sb.nextSectorID++
+	return id, nil
+}
+
+func (sb *SBMock) Scrub(sectorbuilder.SortedPublicSectorInfo) []*sectorbuilder.Fault {
+	sb.lk.Lock()
+	mcopy := make(map[uint64]*sectorState)
+	for k, v := range sb.sectors {
+		mcopy[k] = v
+	}
+	sb.lk.Unlock()
+
+	var out []*sectorbuilder.Fault
+	for sid, ss := range mcopy {
+		ss.lk.Lock()
+		if ss.failed {
+			out = append(out, &sectorbuilder.Fault{
+				SectorID: sid,
+				Err:      fmt.Errorf("mock sector failed"),
+			})
+
+		}
+		ss.lk.Unlock()
+	}
+
+	return out
+}
+
+func (sb *SBMock) GenerateFallbackPoSt(sectorbuilder.SortedPublicSectorInfo, [sectorbuilder.CommLen]byte, []uint64) ([]sectorbuilder.EPostCandidate, []byte, error) {
+	panic("NYI")
+}
+
+func (sb *SBMock) SealPreCommit(ctx context.Context, sid uint64, ticket sectorbuilder.SealTicket, pieces []sectorbuilder.PublicPieceInfo) (sectorbuilder.RawSealPreCommitOutput, error) {
+	sb.lk.Lock()
+	ss, ok := sb.sectors[sid]
+	sb.lk.Unlock()
+	if !ok {
+		return sectorbuilder.RawSealPreCommitOutput{}, xerrors.Errorf("no sector with id %d in sectorbuilder", sid)
+	}
+
+	ss.lk.Lock()
+	defer ss.lk.Unlock()
+
+	ussize := sectorbuilder.UserBytesForSectorSize(sb.sectorSize)
+
+	// TODO: verify pieces in sinfo.pieces match passed in pieces
+
+	var sum uint64
+	for _, p := range pieces {
+		sum += p.Size
+	}
+
+	if sum != ussize {
+		return sectorbuilder.RawSealPreCommitOutput{}, xerrors.Errorf("aggregated piece sizes don't match up: %d != %d", sum, ussize)
+	}
+
+	if ss.state != statePacking {
+		return sectorbuilder.RawSealPreCommitOutput{}, xerrors.Errorf("cannot call pre-seal on sector not in 'packing' state")
+	}
+
+	opFinishWait(ctx)
+
+	ss.state = statePreCommit
+
+	return sectorbuilder.RawSealPreCommitOutput{
+		CommD: randComm(),
+		CommR: randComm(),
+	}, nil
+}
+
+func (sb *SBMock) SealCommit(ctx context.Context, sid uint64, ticket sectorbuilder.SealTicket, seed sectorbuilder.SealSeed, pieces []sectorbuilder.PublicPieceInfo, precommit sectorbuilder.RawSealPreCommitOutput) ([]byte, error) {
+	sb.lk.Lock()
+	ss, ok := sb.sectors[sid]
+	sb.lk.Unlock()
+	if !ok {
+		return nil, xerrors.Errorf("no such sector %d", sid)
+	}
+	ss.lk.Lock()
+	defer ss.lk.Unlock()
+
+	if ss.failed {
+		return nil, xerrors.Errorf("[mock] cannot commit failed sector %d", sid)
+	}
+
+	if ss.state != statePreCommit {
+		return nil, xerrors.Errorf("cannot commit sector that has not been precommitted")
+	}
+
+	opFinishWait(ctx)
+
+	buf := make([]byte, 32)
+	rand.Read(buf)
+	return buf, nil
+}
+
+func (sb *SBMock) GetPath(string, string) (string, error) {
+	panic("nyi")
+}
+
+func (sb *SBMock) WorkerStats() sectorbuilder.WorkerStats {
+	panic("nyi")
+}
+
+func (sb *SBMock) AddWorker(context.Context, sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error) {
+	panic("nyi")
+}
+
+func (sb *SBMock) TaskDone(context.Context, uint64, sectorbuilder.SealRes) error {
+	panic("nyi")
+}
+
+// Test Instrumentation Methods
+
+func (sb *SBMock) FailSector(sid uint64) error {
+	sb.lk.Lock()
+	defer sb.lk.Unlock()
+	ss, ok := sb.sectors[sid]
+	if !ok {
+		return fmt.Errorf("no such sector in sectorbuilder")
+	}
+
+	ss.failed = true
+	return nil
+}
+
+func opFinishWait(ctx context.Context) {
+	val, ok := ctx.Value("opfinish").(chan struct{})
+	if !ok {
+		return
+	}
+	<-val
+}
+
+func AddOpFinish(ctx context.Context) (context.Context, func()) {
+	done := make(chan struct{})
+
+	return context.WithValue(ctx, "opfinish", done), func() {
+		close(done)
+	}
+}
+
+func (sb *SBMock) StageFakeData() (uint64, []sectorbuilder.PublicPieceInfo, error) {
+	usize := sectorbuilder.UserBytesForSectorSize(sb.sectorSize)
+	sid, err := sb.AcquireSectorId()
+	if err != nil {
+		return 0, nil, err
+	}
+
+	buf := make([]byte, usize)
+	rand.Read(buf)
+
+	pi, err := sb.AddPiece(usize, sid, bytes.NewReader(buf), nil)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	return sid, []sectorbuilder.PublicPieceInfo{pi}, nil
+}

storage/sbmock/sbmock_test.go

@@ -0,0 +1,45 @@
+package sbmock
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/filecoin-project/go-sectorbuilder"
+)
+
+func TestOpFinish(t *testing.T) {
+	sb := NewMockSectorBuilder(1, 1024)
+
+	sid, pieces, err := sb.StageFakeData()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, done := AddOpFinish(context.TODO())
+
+	finished := make(chan struct{})
+	go func() {
+		_, err := sb.SealPreCommit(ctx, sid, sectorbuilder.SealTicket{}, pieces)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+
+		close(finished)
+	}()
+
+	select {
+	case <-finished:
+		t.Fatal("should not finish until we tell it to")
+	case <-time.After(time.Second / 2):
+	}
+
+	done()
+
+	select {
+	case <-finished:
+	case <-time.After(time.Second / 2):
+		t.Fatal("should finish after we tell it to")
+	}
+}

storage/sector_states.go

@@ -69,7 +69,7 @@ func (m *Miner) handleUnsealed(ctx context.Context, sector SectorInfo) *sectorUp
 		return sector.upd().fatal(err)
 	}
 
-	rspco, err := m.sb.SealPreCommit(sector.SectorID, *ticket, sector.pieceInfos())
+	rspco, err := m.sb.SealPreCommit(ctx, sector.SectorID, *ticket, sector.pieceInfos())
 	if err != nil {
 		return sector.upd().to(api.SealFailed).error(xerrors.Errorf("seal pre commit failed: %w", err))
 	}
@@ -173,7 +173,7 @@ func (m *Miner) handlePreCommitted(ctx context.Context, sector SectorInfo) *sect
 func (m *Miner) handleCommitting(ctx context.Context, sector SectorInfo) *sectorUpdate {
 	log.Info("scheduling seal proof computation...")
 
-	proof, err := m.sb.SealCommit(sector.SectorID, sector.Ticket.SB(), sector.Seed.SB(), sector.pieceInfos(), sector.rspco())
+	proof, err := m.sb.SealCommit(ctx, sector.SectorID, sector.Ticket.SB(), sector.Seed.SB(), sector.pieceInfos(), sector.rspco())
 	if err != nil {
 		return sector.upd().to(api.SealCommitFailed).error(xerrors.Errorf("computing seal proof failed: %w", err))
 	}

storage/sectorblocks/blocks.go

@@ -47,10 +47,10 @@ type SectorBlocks struct {
 	keyLk sync.Mutex
 }
 
-func NewSectorBlocks(miner *storage.Miner, ds dtypes.MetadataDS, sb *sectorbuilder.SectorBuilder) *SectorBlocks {
+func NewSectorBlocks(miner *storage.Miner, ds dtypes.MetadataDS, sb storage.SectorBuilder) *SectorBlocks {
 	sbc := &SectorBlocks{
 		Miner: miner,
-		sb:    sb,
+		sb:    sb.(*sectorbuilder.SectorBuilder),
 
 		intermediate: blockstore.NewBlockstore(namespace.Wrap(ds, imBlocksPrefix)),