lotus/extern/sector-storage/piece_provider_test.go
2021-05-20 15:25:46 -06:00

125 lines
4.0 KiB
Go

package sectorstorage
import (
"context"
"io/ioutil"
"strings"
"testing"
"time"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-statestore"
specstorage "github.com/filecoin-project/specs-storage/storage"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
ds_sync "github.com/ipfs/go-datastore/sync"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
)
// TestPieceProviderReadPiece verifies that the ReadPiece method works correctly
func TestPieceProviderReadPiece(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
runTest := func(t *testing.T, alreadyUnsealed bool) {
// Set up sector storage manager
storage := newTestStorage(t)
index := stores.NewIndex()
localStore, err := stores.NewLocal(ctx, storage, index, nil)
require.NoError(t, err)
remoteStore := stores.NewRemote(localStore, index, nil, 6000, &stores.DefaultPartialFileHandler{})
dstore := ds_sync.MutexWrap(datastore.NewMapDatastore())
wsts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/worker/calls")))
smsts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/stmgr/calls")))
sealerCfg := SealerConfig{
ParallelFetchLimit: 10,
AllowAddPiece: true,
AllowPreCommit1: true,
AllowPreCommit2: true,
AllowCommit: true,
AllowUnseal: true,
}
mgr, err := New(ctx, localStore, remoteStore, storage, index, sealerCfg, wsts, smsts)
require.NoError(t, err)
// Set up worker
localTasks := []sealtasks.TaskType{
sealtasks.TTAddPiece, sealtasks.TTPreCommit1, sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTFetch,
}
csts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/stmgr/calls")))
// passing a nil executor here mirrors an actual worker setup as it
// will initialize the worker to use the rust proofs lib under the hood
worker := newLocalWorker(nil, WorkerConfig{
TaskTypes: localTasks,
}, remoteStore, localStore, index, mgr, csts)
err = mgr.AddWorker(ctx, worker)
require.NoError(t, err)
// Create piece provider
pp := NewPieceProvider(remoteStore, index, mgr)
// Mock sector
sector := specstorage.SectorRef{
ID: abi.SectorID{
Miner: 1000,
Number: 1,
},
ProofType: abi.RegisteredSealProof_StackedDrg8MiBV1,
}
// Create some data that when padded will be 8MB
pieceData := strings.Repeat("testthis", 127*1024*8)
size := abi.UnpaddedPieceSize(len(pieceData))
pieceInfo, err := mgr.AddPiece(ctx, sector, nil, size, strings.NewReader(pieceData))
require.NoError(t, err)
// pre-commit 1
pieces := []abi.PieceInfo{pieceInfo}
ticket := abi.SealRandomness{9, 9, 9, 9, 9, 9, 9, 9}
preCommit1, err := mgr.SealPreCommit1(ctx, sector, ticket, pieces)
require.NoError(t, err)
// pre-commit 2
sectorCids, err := mgr.SealPreCommit2(ctx, sector, preCommit1)
require.NoError(t, err)
commD := sectorCids.Unsealed
// If we want to test what happens when the data must be unsealed
// (ie there is not an unsealed copy already available)
if !alreadyUnsealed {
// Remove the unsealed copy from local storage
err = localStore.Remove(ctx, sector.ID, storiface.FTUnsealed, false)
require.NoError(t, err)
}
// Read the piece
offset := storiface.UnpaddedByteIndex(0)
require.NoError(t, err)
reader, unsealed, err := pp.ReadPiece(ctx, sector, offset, size, ticket, commD)
require.NoError(t, err)
requiresUnseal := !alreadyUnsealed
require.Equal(t, requiresUnseal, unsealed)
defer func() { _ = reader.Close() }()
// Make sure the input matches the output
readData, err := ioutil.ReadAll(reader)
require.NoError(t, err)
require.Equal(t, pieceData, string(readData))
}
t.Run("already unsealed", func(t *testing.T) {
runTest(t, true)
})
t.Run("requires unseal", func(t *testing.T) {
runTest(t, false)
})
}