cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #1510 from filecoin-project/feat/fsm-updates

Browse Source 
FSM Updates
This commit is contained in:
Łukasz Magiera 2020-04-06 23:11:46 +02:00 committed by GitHub
commit 4ee8254133
27 changed files with 517 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
api/api_full.go
View File

@ -127,6 +127,7 @@ type FullNode interface {
StateMinerPostState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*miner.PoStState, error)
StateMinerSectorSize(context.Context, address.Address, types.TipSetKey) (abi.SectorSize, error)
StateMinerFaults(context.Context, address.Address, types.TipSetKey) ([]abi.SectorNumber, error)
StateSectorPreCommitInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error)
StatePledgeCollateral(context.Context, types.TipSetKey) (types.BigInt, error)
StateWaitMsg(context.Context, cid.Cid) (*MsgLookup, error)
StateSearchMsg(context.Context, cid.Cid) (*MsgLookup, error)

87
api/api_storage.go
View File

@ -15,79 +15,34 @@ import (
"github.com/filecoin-project/sector-storage/stores"
)
// alias because cbor-gen doesn't like non-alias types
type SectorState = uint64
type SectorState string
const (
UndefinedSectorState SectorState = iota
UndefinedSectorState SectorState = ""
// happy path
Empty
Packing // sector not in sealStore, and not on chain
Unsealed // sealing / queued
PreCommitting // on chain pre-commit
WaitSeed // waiting for seed
Committing
CommitWait // waiting for message to land on chain
FinalizeSector
Proving
_ // reserved
_
_
// recovery handling
// Reseal
_
_
_
_
_
_
_
Empty SectorState = "Empty"
Packing SectorState = "Packing" // sector not in sealStore, and not on chain
PreCommit1 SectorState = "PreCommit1" // do PreCommit1
PreCommit2 SectorState = "PreCommit2" // do PreCommit1
PreCommitting SectorState = "PreCommitting" // on chain pre-commit
WaitSeed SectorState = "WaitSeed" // waiting for seed
Committing SectorState = "Committing"
CommitWait SectorState = "CommitWait" // waiting for message to land on chain
FinalizeSector SectorState = "FinalizeSector"
Proving SectorState = "Proving"
// error modes
FailedUnrecoverable
SealFailed
PreCommitFailed
SealCommitFailed
CommitFailed
PackingFailed
_
_
_
Faulty // sector is corrupted or gone for some reason
FaultReported // sector has been declared as a fault on chain
FaultedFinal // fault declared on chain
FailedUnrecoverable SectorState = "FailedUnrecoverable"
SealFailed SectorState = "SealFailed"
PreCommitFailed SectorState = "PreCommitFailed"
ComputeProofFailed SectorState = "ComputeProofFailed"
CommitFailed SectorState = "CommitFailed"
PackingFailed SectorState = "PackingFailed"
Faulty SectorState = "Faulty" // sector is corrupted or gone for some reason
FaultReported SectorState = "FaultReported" // sector has been declared as a fault on chain
FaultedFinal SectorState = "FaultedFinal" // fault declared on chain
)
var SectorStates = []string{
UndefinedSectorState: "UndefinedSectorState",
Empty: "Empty",
Packing: "Packing",
Unsealed: "Unsealed",
PreCommitting: "PreCommitting",
WaitSeed: "WaitSeed",
Committing: "Committing",
CommitWait: "CommitWait",
FinalizeSector: "FinalizeSector",
Proving: "Proving",
SealFailed: "SealFailed",
PreCommitFailed: "PreCommitFailed",
SealCommitFailed: "SealCommitFailed",
CommitFailed: "CommitFailed",
PackingFailed: "PackingFailed",
FailedUnrecoverable: "FailedUnrecoverable",
Faulty: "Faulty",
FaultReported: "FaultReported",
FaultedFinal: "FaultedFinal",
}
// StorageMiner is a low-level interface to the Filecoin network storage miner node
type StorageMiner interface {
Common

63
api/apistruct/struct.go
View File

@ -112,35 +112,36 @@ type FullNodeStruct struct {
ClientRetrieve func(ctx context.Context, order api.RetrievalOrder, ref api.FileRef) error `perm:"admin"`
ClientQueryAsk func(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.SignedStorageAsk, error) `perm:"read"`
StateNetworkName func(context.Context) (dtypes.NetworkName, error) `perm:"read"`
StateMinerSectors func(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error) `perm:"read"`
StateMinerProvingSet func(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error) `perm:"read"`
StateMinerPower func(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error) `perm:"read"`
StateMinerWorker func(context.Context, address.Address, types.TipSetKey) (address.Address, error) `perm:"read"`
StateMinerPeerID func(ctx context.Context, m address.Address, tsk types.TipSetKey) (peer.ID, error) `perm:"read"`
StateMinerPostState func(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*miner.PoStState, error) `perm:"read"`
StateMinerSectorSize func(context.Context, address.Address, types.TipSetKey) (abi.SectorSize, error) `perm:"read"`
StateMinerFaults func(context.Context, address.Address, types.TipSetKey) ([]abi.SectorNumber, error) `perm:"read"`
StateCall func(context.Context, *types.Message, types.TipSetKey) (*api.InvocResult, error) `perm:"read"`
StateReplay func(context.Context, types.TipSetKey, cid.Cid) (*api.InvocResult, error) `perm:"read"`
StateGetActor func(context.Context, address.Address, types.TipSetKey) (*types.Actor, error) `perm:"read"`
StateReadState func(context.Context, *types.Actor, types.TipSetKey) (*api.ActorState, error) `perm:"read"`
StatePledgeCollateral func(context.Context, types.TipSetKey) (types.BigInt, error) `perm:"read"`
StateWaitMsg func(context.Context, cid.Cid) (*api.MsgLookup, error) `perm:"read"`
StateSearchMsg func(context.Context, cid.Cid) (*api.MsgLookup, error) `perm:"read"`
StateListMiners func(context.Context, types.TipSetKey) ([]address.Address, error) `perm:"read"`
StateListActors func(context.Context, types.TipSetKey) ([]address.Address, error) `perm:"read"`
StateMarketBalance func(context.Context, address.Address, types.TipSetKey) (api.MarketBalance, error) `perm:"read"`
StateMarketParticipants func(context.Context, types.TipSetKey) (map[string]api.MarketBalance, error) `perm:"read"`
StateMarketDeals func(context.Context, types.TipSetKey) (map[string]api.MarketDeal, error) `perm:"read"`
StateMarketStorageDeal func(context.Context, abi.DealID, types.TipSetKey) (*api.MarketDeal, error) `perm:"read"`
StateLookupID func(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) `perm:"read"`
StateChangedActors func(context.Context, cid.Cid, cid.Cid) (map[string]types.Actor, error) `perm:"read"`
StateGetReceipt func(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error) `perm:"read"`
StateMinerSectorCount func(context.Context, address.Address, types.TipSetKey) (api.MinerSectors, error) `perm:"read"`
StateListMessages func(ctx context.Context, match *types.Message, tsk types.TipSetKey, toht abi.ChainEpoch) ([]cid.Cid, error) `perm:"read"`
StateListRewards func(context.Context, address.Address, types.TipSetKey) ([]reward.Reward, error) `perm:"read"`
StateCompute func(context.Context, abi.ChainEpoch, []*types.Message, types.TipSetKey) (*api.ComputeStateOutput, error) `perm:"read"`
StateNetworkName func(context.Context) (dtypes.NetworkName, error) `perm:"read"`
StateMinerSectors func(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error) `perm:"read"`
StateMinerProvingSet func(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error) `perm:"read"`
StateMinerPower func(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error) `perm:"read"`
StateMinerWorker func(context.Context, address.Address, types.TipSetKey) (address.Address, error) `perm:"read"`
StateMinerPeerID func(ctx context.Context, m address.Address, tsk types.TipSetKey) (peer.ID, error) `perm:"read"`
StateMinerPostState func(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*miner.PoStState, error) `perm:"read"`
StateMinerSectorSize func(context.Context, address.Address, types.TipSetKey) (abi.SectorSize, error) `perm:"read"`
StateMinerFaults func(context.Context, address.Address, types.TipSetKey) ([]abi.SectorNumber, error) `perm:"read"`
StateSectorPreCommitInfo func(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error) `perm:"read"`
StateCall func(context.Context, *types.Message, types.TipSetKey) (*api.InvocResult, error) `perm:"read"`
StateReplay func(context.Context, types.TipSetKey, cid.Cid) (*api.InvocResult, error) `perm:"read"`
StateGetActor func(context.Context, address.Address, types.TipSetKey) (*types.Actor, error) `perm:"read"`
StateReadState func(context.Context, *types.Actor, types.TipSetKey) (*api.ActorState, error) `perm:"read"`
StatePledgeCollateral func(context.Context, types.TipSetKey) (types.BigInt, error) `perm:"read"`
StateWaitMsg func(context.Context, cid.Cid) (*api.MsgLookup, error) `perm:"read"`
StateSearchMsg func(context.Context, cid.Cid) (*api.MsgLookup, error) `perm:"read"`
StateListMiners func(context.Context, types.TipSetKey) ([]address.Address, error) `perm:"read"`
StateListActors func(context.Context, types.TipSetKey) ([]address.Address, error) `perm:"read"`
StateMarketBalance func(context.Context, address.Address, types.TipSetKey) (api.MarketBalance, error) `perm:"read"`
StateMarketParticipants func(context.Context, types.TipSetKey) (map[string]api.MarketBalance, error) `perm:"read"`
StateMarketDeals func(context.Context, types.TipSetKey) (map[string]api.MarketDeal, error) `perm:"read"`
StateMarketStorageDeal func(context.Context, abi.DealID, types.TipSetKey) (*api.MarketDeal, error) `perm:"read"`
StateLookupID func(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) `perm:"read"`
StateChangedActors func(context.Context, cid.Cid, cid.Cid) (map[string]types.Actor, error) `perm:"read"`
StateGetReceipt func(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error) `perm:"read"`
StateMinerSectorCount func(context.Context, address.Address, types.TipSetKey) (api.MinerSectors, error) `perm:"read"`
StateListMessages func(ctx context.Context, match *types.Message, tsk types.TipSetKey, toht abi.ChainEpoch) ([]cid.Cid, error) `perm:"read"`
StateListRewards func(context.Context, address.Address, types.TipSetKey) ([]reward.Reward, error) `perm:"read"`
StateCompute func(context.Context, abi.ChainEpoch, []*types.Message, types.TipSetKey) (*api.ComputeStateOutput, error) `perm:"read"`
MsigGetAvailableBalance func(context.Context, address.Address, types.TipSetKey) (types.BigInt, error) `perm:"read"`
@ -512,6 +513,10 @@ func (c *FullNodeStruct) StateMinerFaults(ctx context.Context, actor address.Add
return c.Internal.StateMinerFaults(ctx, actor, tsk)
}
func (c *FullNodeStruct) StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error) {
return c.Internal.StateSectorPreCommitInfo(ctx, maddr, n, tsk)
}
func (c *FullNodeStruct) StateCall(ctx context.Context, msg *types.Message, tsk types.TipSetKey) (*api.InvocResult, error) {
return c.Internal.StateCall(ctx, msg, tsk)
}

4
build/params_shared.go
View File

@ -60,10 +60,10 @@ const FallbackPoStConfidence = 6
const SealRandomnessLookback = Finality
// Epochs
const SealRandomnessLookbackLimit = SealRandomnessLookback + 2000
const SealRandomnessLookbackLimit = SealRandomnessLookback + 2000 // TODO: Get from spec specs-actors
// Maximum lookback that randomness can be sourced from for a seal proof submission
const MaxSealLookback = SealRandomnessLookbackLimit + 2000
const MaxSealLookback = SealRandomnessLookbackLimit + 2000 // TODO: Get from specs-actors
// /////
// Mining

2
chain/blocksync/cbor_gen.go
View File

@ -7,7 +7,7 @@ import (
"io"
"github.com/filecoin-project/lotus/chain/types"
"github.com/ipfs/go-cid"
cid "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)

19
chain/stmgr/utils.go
View File

@ -2,7 +2,6 @@ package stmgr
import (
"context"
amt "github.com/filecoin-project/go-amt-ipld/v2"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/abi/big"
@ -145,6 +144,24 @@ func SectorSetSizes(ctx context.Context, sm *StateManager, maddr address.Address
}, nil
}
func PreCommitInfo(ctx context.Context, sm *StateManager, maddr address.Address, sid abi.SectorNumber, ts *types.TipSet) (miner.SectorPreCommitOnChainInfo, error) {
var mas miner.State
_, err := sm.LoadActorState(ctx, maddr, &mas, ts)
if err != nil {
return miner.SectorPreCommitOnChainInfo{}, xerrors.Errorf("(get sset) failed to load miner actor state: %w", err)
}
i, ok, err := mas.GetPrecommittedSector(sm.cs.Store(ctx), sid)
if err != nil {
return miner.SectorPreCommitOnChainInfo{}, err
}
if !ok {
return miner.SectorPreCommitOnChainInfo{}, xerrors.New("precommit not found")
}
return *i, nil
}
func GetMinerProvingSet(ctx context.Context, sm *StateManager, ts *types.TipSet, maddr address.Address) ([]*api.ChainSectorInfo, error) {
return getMinerProvingSetRaw(ctx, sm, ts.ParentState(), maddr)
}

2
chain/types/cbor_gen.go
View File

@ -9,7 +9,7 @@ import (
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/crypto"
"github.com/filecoin-project/specs-actors/actors/runtime/exitcode"
"github.com/ipfs/go-cid"
cid "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)

2
cli/chain.go
View File

@ -433,6 +433,8 @@ var chainGetCmd = &cli.Command{
Note:
You can use special path elements to traverse through some data structures:
- /ipfs/[cid]/@H:elem - get 'elem' from hamt
- /ipfs/[cid]/@Hi:123 - get varint elem 123 from hamt
- /ipfs/[cid]/@Hu:123 - get uvarint elem 123 from hamt
- /ipfs/[cid]/@Ha:t01 - get element under Addr(t01).Bytes
- /ipfs/[cid]/@A:10 - get 10th amt element
`,

6
cmd/lotus-storage-miner/info.go
View File

@ -127,13 +127,13 @@ var infoCmd = &cli.Command{
},
}
func sectorsInfo(ctx context.Context, napi api.StorageMiner) (map[string]int, error) {
func sectorsInfo(ctx context.Context, napi api.StorageMiner) (map[api.SectorState]int, error) {
sectors, err := napi.SectorsList(ctx)
if err != nil {
return nil, err
}
out := map[string]int{
out := map[api.SectorState]int{
"Total": len(sectors),
}
for _, s := range sectors {
@ -142,7 +142,7 @@ func sectorsInfo(ctx context.Context, napi api.StorageMiner) (map[string]int, er
return nil, err
}
out[api.SectorStates[st.State]]++
out[st.State]++
}
return out, nil

13
cmd/lotus-storage-miner/sectors.go
View File

@ -77,7 +77,7 @@ var sectorsStatusCmd = &cli.Command{
}
fmt.Printf("SectorID:\t%d\n", status.SectorID)
fmt.Printf("Status:\t%s\n", api.SectorStates[status.State])
fmt.Printf("Status:\t%s\n", status.State)
fmt.Printf("CommD:\t\t%x\n", status.CommD)
fmt.Printf("CommR:\t\t%x\n", status.CommR)
fmt.Printf("Ticket:\t\t%x\n", status.Ticket.Value)
@ -169,7 +169,7 @@ var sectorsListCmd = &cli.Command{
fmt.Fprintf(w, "%d: %s\tsSet: %s\tpSet: %s\ttktH: %d\tseedH: %d\tdeals: %v\n",
s,
api.SectorStates[st.State],
st.State,
yesno(inSSet),
yesno(inPSet),
st.Ticket.Epoch,
@ -236,14 +236,7 @@ var sectorsUpdateCmd = &cli.Command{
return xerrors.Errorf("could not parse sector ID: %w", err)
}
var st api.SectorState
for i, s := range api.SectorStates {
if cctx.Args().Get(1) == s {
st = api.SectorState(i)
}
}
return nodeApi.SectorsUpdate(ctx, abi.SectorNumber(id), st)
return nodeApi.SectorsUpdate(ctx, abi.SectorNumber(id), api.SectorState(cctx.Args().Get(1)))
},
}

1
node/builder.go
View File

@ -274,6 +274,7 @@ func Online() Option {
Override(new(stores.LocalStorage), From(new(repo.LockedRepo))),
Override(new(sealing.SectorIDCounter), modules.SectorIDCounter),
Override(new(*sectorstorage.Manager), modules.SectorStorage),
Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
Override(new(sectorstorage.SectorManager), From(new(*sectorstorage.Manager))),
Override(new(storage2.Prover), From(new(sectorstorage.SectorManager))),

2
node/hello/cbor_gen.go
View File

@ -7,7 +7,7 @@ import (
"io"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/ipfs/go-cid"
cid "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)

11
node/impl/full/chain.go
View File

@ -285,6 +285,17 @@ func resolveOnce(bs blockstore.Blockstore) func(ctx context.Context, ds ipld.Nod
names[0] = "@H:" + ik.Key()
}
if strings.HasPrefix(names[0], "@Hu:") {
i, err := strconv.ParseUint(names[0][4:], 10, 64)
if err != nil {
return nil, nil, xerrors.Errorf("parsing int64: %w", err)
}
ik := adt.UIntKey(i)
names[0] = "@H:" + ik.Key()
}
if strings.HasPrefix(names[0], "@H:") {
cst := cbor.NewCborStore(bs)

8
node/impl/full/state.go
View File

@ -514,6 +514,14 @@ func (a *StateAPI) StateMinerSectorCount(ctx context.Context, addr address.Addre
return stmgr.SectorSetSizes(ctx, a.StateManager, addr, ts)
}
func (a *StateAPI) StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
return miner.SectorPreCommitOnChainInfo{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
return stmgr.PreCommitInfo(ctx, a.StateManager, maddr, n, ts)
}
func (a *StateAPI) StateListMessages(ctx context.Context, match *types.Message, tsk types.TipSetKey, toheight abi.ChainEpoch) ([]cid.Cid, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {

4
node/modules/storageminer.go
View File

@ -123,7 +123,7 @@ func SectorIDCounter(ds dtypes.MetadataDS) sealing.SectorIDCounter {
return &sidsc{sc}
}
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, tktFn sealing.TicketFn) (*storage.Miner, error) {
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, tktFn sealing.TicketFn) (*storage.Miner, error) {
maddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
@ -143,7 +143,7 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h
fps := storage.NewFPoStScheduler(api, sealer, maddr, worker, ppt)
sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, tktFn)
sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, sc, verif, tktFn)
if err != nil {
return nil, err
}

1
node/node_test.go
View File

@ -391,6 +391,7 @@ func mockSbBuilder(t *testing.T, nFull int, storage []int) ([]test.TestNode, []t
node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
return mock.NewMockSectorMgr(5, build.SectorSizes[0]), nil
}),
node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
node.Unset(new(*sectorstorage.Manager)),
))
}

12
storage/miner.go
View File

@ -3,11 +3,9 @@ package storage
import (
"context"
"errors"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"time"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/specs-storage/storage"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
logging "github.com/ipfs/go-log/v2"
@ -15,6 +13,8 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/sector-storage"
"github.com/filecoin-project/sector-storage/ffiwrapper"
"github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/builtin/miner"
@ -26,6 +26,7 @@ import (
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/storage/sealing"
)
@ -38,6 +39,7 @@ type Miner struct {
ds datastore.Batching
tktFn sealing.TicketFn
sc sealing.SectorIDCounter
verif ffiwrapper.Verifier
maddr address.Address
worker address.Address
@ -53,6 +55,7 @@ type storageMinerApi interface {
StateMinerSectors(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error)
StateMinerProvingSet(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error)
StateMinerSectorSize(context.Context, address.Address, types.TipSetKey) (abi.SectorSize, error)
StateSectorPreCommitInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error)
StateWaitMsg(context.Context, cid.Cid) (*api.MsgLookup, error) // TODO: removeme eventually
StateGetActor(ctx context.Context, actor address.Address, ts types.TipSetKey) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
@ -74,7 +77,7 @@ type storageMinerApi interface {
WalletHas(context.Context, address.Address) (bool, error)
}
func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, ds datastore.Batching, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, tktFn sealing.TicketFn) (*Miner, error) {
func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, ds datastore.Batching, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, tktFn sealing.TicketFn) (*Miner, error) {
m := &Miner{
api: api,
h: h,
@ -82,6 +85,7 @@ func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, d
ds: ds,
tktFn: tktFn,
sc: sc,
verif: verif,
maddr: maddr,
worker: worker,
@ -96,7 +100,7 @@ func (m *Miner) Run(ctx context.Context) error {
}
evts := events.NewEvents(ctx, m.api)
m.sealing = sealing.New(m.api, evts, m.maddr, m.worker, m.ds, m.sealer, m.sc, m.tktFn)
m.sealing = sealing.New(m.api, evts, m.maddr, m.worker, m.ds, m.sealer, m.sc, m.verif, m.tktFn)
go m.sealing.Run(ctx)

148
storage/sealing/cbor_gen.go
View File

@ -6,6 +6,7 @@ import (
"fmt"
"io"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/specs-actors/actors/abi"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
@ -177,11 +178,11 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{175}); err != nil {
if _, err := w.Write([]byte{177}); err != nil {
return err
}
// t.State (uint64) (uint64)
// t.State (api.SectorState) (string)
if len("State") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"State\" was too long")
}
@ -193,7 +194,14 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
return err
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.State))); err != nil {
if len(t.State) > cbg.MaxLength {
return xerrors.Errorf("Value in field t.State was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len(t.State)))); err != nil {
return err
}
if _, err := w.Write([]byte(t.State)); err != nil {
return err
}
@ -276,6 +284,45 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
}
}
// t.Ticket (api.SealTicket) (struct)
if len("Ticket") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"Ticket\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("Ticket")))); err != nil {
return err
}
if _, err := w.Write([]byte("Ticket")); err != nil {
return err
}
if err := t.Ticket.MarshalCBOR(w); err != nil {
return err
}
// t.PreCommit1Out (storage.PreCommit1Out) (slice)
if len("PreCommit1Out") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"PreCommit1Out\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("PreCommit1Out")))); err != nil {
return err
}
if _, err := w.Write([]byte("PreCommit1Out")); err != nil {
return err
}
if len(t.PreCommit1Out) > cbg.ByteArrayMaxLen {
return xerrors.Errorf("Byte array in field t.PreCommit1Out was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajByteString, uint64(len(t.PreCommit1Out)))); err != nil {
return err
}
if _, err := w.Write(t.PreCommit1Out); err != nil {
return err
}
// t.CommD (cid.Cid) (struct)
if len("CommD") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"CommD\" was too long")
@ -343,22 +390,6 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
return err
}
// t.Ticket (api.SealTicket) (struct)
if len("Ticket") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"Ticket\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("Ticket")))); err != nil {
return err
}
if _, err := w.Write([]byte("Ticket")); err != nil {
return err
}
if err := t.Ticket.MarshalCBOR(w); err != nil {
return err
}
// t.PreCommitMessage (cid.Cid) (struct)
if len("PreCommitMessage") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"PreCommitMessage\" was too long")
@ -419,6 +450,22 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
}
}
// t.InvalidProofs (uint64) (uint64)
if len("InvalidProofs") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"InvalidProofs\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("InvalidProofs")))); err != nil {
return err
}
if _, err := w.Write([]byte("InvalidProofs")); err != nil {
return err
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.InvalidProofs))); err != nil {
return err
}
// t.FaultReportMsg (cid.Cid) (struct)
if len("FaultReportMsg") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"FaultReportMsg\" was too long")
@ -521,20 +568,16 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) error {
}
switch name {
// t.State (uint64) (uint64)
// t.State (api.SectorState) (string)
case "State":
{
maj, extra, err = cbg.CborReadHeader(br)
sval, err := cbg.ReadString(br)
if err != nil {
return err
}
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.State = uint64(extra)
t.State = api.SectorState(sval)
}
// t.SectorID (abi.SectorNumber) (uint64)
case "SectorID":
@ -620,6 +663,34 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) error {
t.Pieces[i] = v
}
// t.Ticket (api.SealTicket) (struct)
case "Ticket":
{
if err := t.Ticket.UnmarshalCBOR(br); err != nil {
return xerrors.Errorf("unmarshaling t.Ticket: %w", err)
}
}
// t.PreCommit1Out (storage.PreCommit1Out) (slice)
case "PreCommit1Out":
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > cbg.ByteArrayMaxLen {
return fmt.Errorf("t.PreCommit1Out: byte array too large (%d)", extra)
}
if maj != cbg.MajByteString {
return fmt.Errorf("expected byte array")
}
t.PreCommit1Out = make([]byte, extra)
if _, err := io.ReadFull(br, t.PreCommit1Out); err != nil {
return err
}
// t.CommD (cid.Cid) (struct)
case "CommD":
@ -688,16 +759,6 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) error {
if _, err := io.ReadFull(br, t.Proof); err != nil {
return err
}
// t.Ticket (api.SealTicket) (struct)
case "Ticket":
{
if err := t.Ticket.UnmarshalCBOR(br); err != nil {
return xerrors.Errorf("unmarshaling t.Ticket: %w", err)
}
}
// t.PreCommitMessage (cid.Cid) (struct)
case "PreCommitMessage":
@ -757,6 +818,21 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) error {
t.CommitMessage = &c
}
}
// t.InvalidProofs (uint64) (uint64)
case "InvalidProofs":
{
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.InvalidProofs = uint64(extra)
}
// t.FaultReportMsg (cid.Cid) (struct)
case "FaultReportMsg":

77
storage/sealing/checks.go
View File

@ -9,8 +9,12 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/sector-storage/ffiwrapper"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/builtin"
"github.com/filecoin-project/specs-actors/actors/builtin/market"
"github.com/filecoin-project/specs-actors/actors/builtin/miner"
"github.com/filecoin-project/specs-actors/actors/crypto"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
@ -29,6 +33,9 @@ type ErrExpiredDeals struct{ error }
type ErrBadCommD struct{ error }
type ErrExpiredTicket struct{ error }
type ErrBadSeed struct{ error }
type ErrInvalidProof struct{ error }
// checkPieces validates that:
// - Each piece han a corresponding on chain deal
// - Piece commitments match with on chain deals
@ -69,9 +76,9 @@ func checkPieces(ctx context.Context, si SectorInfo, api sealingApi) error {
return nil
}
// checkSeal checks that data commitment generated in the sealing process
// checkPrecommit checks that data commitment generated in the sealing process
// matches pieces, and that the seal ticket isn't expired
func checkSeal(ctx context.Context, maddr address.Address, si SectorInfo, api sealingApi) (err error) {
func checkPrecommit(ctx context.Context, maddr address.Address, si SectorInfo, api sealingApi) (err error) {
head, err := api.ChainHead(ctx)
if err != nil {
return &ErrApi{xerrors.Errorf("getting chain head: %w", err)}
@ -116,5 +123,69 @@ func checkSeal(ctx context.Context, maddr address.Address, si SectorInfo, api se
}
return nil
}
func (m *Sealing) checkCommit(ctx context.Context, si SectorInfo, proof []byte) (err error) {
head, err := m.api.ChainHead(ctx)
if err != nil {
return &ErrApi{xerrors.Errorf("getting chain head: %w", err)}
}
if si.Seed.Epoch == 0 {
return &ErrBadSeed{xerrors.Errorf("seed epoch was not set")}
}
pci, err := m.api.StateSectorPreCommitInfo(ctx, m.maddr, si.SectorID, types.EmptyTSK)
if err != nil {
return xerrors.Errorf("getting precommit info: %w", err)
}
if pci.PreCommitEpoch+miner.PreCommitChallengeDelay != si.Seed.Epoch {
return &ErrBadSeed{xerrors.Errorf("seed epoch doesn't match on chain info: %d != %d", pci.PreCommitEpoch+miner.PreCommitChallengeDelay, si.Seed.Epoch)}
}
seed, err := m.api.ChainGetRandomness(ctx, head.Key(), crypto.DomainSeparationTag_InteractiveSealChallengeSeed, si.Seed.Epoch, nil)
if err != nil {
return &ErrApi{xerrors.Errorf("failed to get randomness for computing seal proof: %w", err)}
}
if string(seed) != string(si.Seed.Value) {
return &ErrBadSeed{xerrors.Errorf("seed has changed")}
}
ss, err := m.api.StateMinerSectorSize(ctx, m.maddr, head.Key())
if err != nil {
return &ErrApi{err}
}
_, spt, err := ffiwrapper.ProofTypeFromSectorSize(ss)
if err != nil {
return err
}
if *si.CommR != pci.Info.SealedCID {
log.Warn("on-chain sealed CID doesn't match!")
}
ok, err := m.verif.VerifySeal(abi.SealVerifyInfo{
SectorID: m.minerSector(si.SectorID),
OnChain: abi.OnChainSealVerifyInfo{
SealedCID: pci.Info.SealedCID,
InteractiveEpoch: si.Seed.Epoch,
RegisteredProof: spt,
Proof: proof,
SectorNumber: si.SectorID,
SealRandEpoch: si.Ticket.Epoch,
},
Randomness: si.Ticket.Value,
InteractiveRandomness: si.Seed.Value,
UnsealedCID: *si.CommD,
})
if err != nil {
return xerrors.Errorf("verify seal: %w", err)
}
if !ok {
return &ErrInvalidProof{xerrors.New("invalid proof (compute error?)")}
}
return nil
}

85
storage/sealing/fsm.go
View File

@ -32,22 +32,27 @@ func (m *Sealing) Plan(events []statemachine.Event, user interface{}) (interface
}, uint64(len(events)), nil // TODO: This processed event count is not very correct
}
var fsmPlanners = []func(events []statemachine.Event, state *SectorInfo) error{
var fsmPlanners = map[api.SectorState]func(events []statemachine.Event, state *SectorInfo) error{
api.UndefinedSectorState: planOne(on(SectorStart{}, api.Packing)),
api.Packing: planOne(on(SectorPacked{}, api.Unsealed)),
api.Unsealed: planOne(
on(SectorSealed{}, api.PreCommitting),
on(SectorSealFailed{}, api.SealFailed),
api.Packing: planOne(on(SectorPacked{}, api.PreCommit1)),
api.PreCommit1: planOne(
on(SectorPreCommit1{}, api.PreCommit2),
on(SectorSealPreCommitFailed{}, api.SealFailed),
on(SectorPackingFailed{}, api.PackingFailed),
),
api.PreCommit2: planOne(
on(SectorPreCommit2{}, api.PreCommitting),
on(SectorSealPreCommitFailed{}, api.SealFailed),
on(SectorPackingFailed{}, api.PackingFailed),
),
api.PreCommitting: planOne(
on(SectorSealFailed{}, api.SealFailed),
on(SectorSealPreCommitFailed{}, api.SealFailed),
on(SectorPreCommitted{}, api.WaitSeed),
on(SectorPreCommitFailed{}, api.PreCommitFailed),
on(SectorChainPreCommitFailed{}, api.PreCommitFailed),
),
api.WaitSeed: planOne(
on(SectorSeedReady{}, api.Committing),
on(SectorPreCommitFailed{}, api.PreCommitFailed),
on(SectorChainPreCommitFailed{}, api.PreCommitFailed),
),
api.Committing: planCommitting,
api.CommitWait: planOne(
@ -65,12 +70,21 @@ var fsmPlanners = []func(events []statemachine.Event, state *SectorInfo) error{
),
api.SealFailed: planOne(
on(SectorRetrySeal{}, api.Unsealed),
on(SectorRetrySeal{}, api.PreCommit1),
),
api.PreCommitFailed: planOne(
on(SectorRetryPreCommit{}, api.PreCommitting),
on(SectorRetryWaitSeed{}, api.WaitSeed),
on(SectorSealFailed{}, api.SealFailed),
on(SectorSealPreCommitFailed{}, api.SealFailed),
),
api.ComputeProofFailed: planOne(
on(SectorRetryComputeProof{}, api.Committing),
),
api.CommitFailed: planOne(
on(SectorSealPreCommitFailed{}, api.SealFailed),
on(SectorRetryWaitSeed{}, api.WaitSeed),
on(SectorRetryComputeProof{}, api.Committing),
on(SectorRetryInvalidProof{}, api.Committing),
),
api.Faulty: planOne(
@ -105,11 +119,11 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
p := fsmPlanners[state.State]
if p == nil {
return nil, xerrors.Errorf("planner for state %s not found", api.SectorStates[state.State])
return nil, xerrors.Errorf("planner for state %s not found", state.State)
}
if err := p(events, state); err != nil {
return nil, xerrors.Errorf("running planner for state %s failed: %w", api.SectorStates[state.State], err)
return nil, xerrors.Errorf("running planner for state %s failed: %w", state.State, err)
}
/////
@ -123,16 +137,21 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
*<- Packing <- incoming
| |
| v
*<- Unsealed <--> SealFailed
| |
| v
* PreCommitting <--> PreCommitFailed
| | ^
| v |
*<- WaitSeed ----------/
| |||
| vvv v--> SealCommitFailed
*<- Committing
*<- PreCommit1 <--> SealFailed
| | ^^^
| v |||
*<- PreCommit2 -------/||
| | ||
| v /-------/|
* PreCommitting <-----+---> PreCommitFailed
| | | ^
| v | |
*<- WaitSeed -----------+-----/
| ||| ^ |
| ||| \--------*-----/
| ||| |
| vvv v----+----> ComputeProofFailed
*<- Committing |
| | ^--> CommitFailed
| v ^
*<- CommitWait ---/
@ -153,8 +172,10 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
// Happy path
case api.Packing:
return m.handlePacking, nil
case api.Unsealed:
return m.handleUnsealed, nil
case api.PreCommit1:
return m.handlePreCommit1, nil
case api.PreCommit2:
return m.handlePreCommit2, nil
case api.PreCommitting:
return m.handlePreCommitting, nil
case api.WaitSeed:
@ -174,10 +195,10 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
return m.handleSealFailed, nil
case api.PreCommitFailed:
return m.handlePreCommitFailed, nil
case api.SealCommitFailed:
log.Warnf("sector %d entered unimplemented state 'SealCommitFailed'", state.SectorID)
case api.ComputeProofFailed:
return m.handleComputeProofFailed, nil
case api.CommitFailed:
log.Warnf("sector %d entered unimplemented state 'CommitFailed'", state.SectorID)
return m.handleCommitFailed, nil
// Faults
case api.Faulty:
@ -217,8 +238,8 @@ func planCommitting(events []statemachine.Event, state *SectorInfo) error {
state.State = api.Committing
return nil
case SectorComputeProofFailed:
state.State = api.SealCommitFailed
case SectorSealFailed:
state.State = api.ComputeProofFailed
case SectorSealPreCommitFailed:
state.State = api.CommitFailed
case SectorCommitFailed:
state.State = api.CommitFailed
@ -251,7 +272,7 @@ func (m *Sealing) ForceSectorState(ctx context.Context, id abi.SectorNumber, sta
}
func final(events []statemachine.Event, state *SectorInfo) error {
return xerrors.Errorf("didn't expect any events in state %s, got %+v", api.SectorStates[state.State], events)
return xerrors.Errorf("didn't expect any events in state %s, got %+v", state.State, events)
}
func on(mut mutator, next api.SectorState) func() (mutator, api.SectorState) {
@ -269,7 +290,7 @@ func planOne(ts ...func() (mut mutator, next api.SectorState)) func(events []sta
return nil
}
}
return xerrors.Errorf("planner for state %s only has a plan for a single event only, got %+v", api.SectorStates[state.State], events)
return xerrors.Errorf("planner for state %s only has a plan for a single event only, got %+v", state.State, events)
}
if gm, ok := events[0].User.(globalMutator); ok {
@ -293,6 +314,6 @@ func planOne(ts ...func() (mut mutator, next api.SectorState)) func(events []sta
return nil
}
return xerrors.Errorf("planner for state %s received unexpected event %T (%+v)", api.SectorStates[state.State], events[0].User, events[0])
return xerrors.Errorf("planner for state %s received unexpected event %T (%+v)", state.State, events[0].User, events[0])
}
}

45
storage/sealing/fsm_events.go
View File

@ -2,6 +2,7 @@ package sealing
import (
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage"
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
@ -70,29 +71,39 @@ type SectorPackingFailed struct{ error }
func (evt SectorPackingFailed) apply(*SectorInfo) {}
type SectorSealed struct {
Sealed cid.Cid
Unsealed cid.Cid
Ticket api.SealTicket
type SectorPreCommit1 struct {
PreCommit1Out storage.PreCommit1Out
Ticket api.SealTicket
}
func (evt SectorSealed) apply(state *SectorInfo) {
func (evt SectorPreCommit1) apply(state *SectorInfo) {
state.PreCommit1Out = evt.PreCommit1Out
state.Ticket = evt.Ticket
}
type SectorPreCommit2 struct {
Sealed cid.Cid
Unsealed cid.Cid
}
func (evt SectorPreCommit2) apply(state *SectorInfo) {
commd := evt.Unsealed
state.CommD = &commd
commr := evt.Sealed
state.CommR = &commr
state.Ticket = evt.Ticket
}
type SectorSealFailed struct{ error }
type SectorSealPreCommitFailed struct{ error }
func (evt SectorSealFailed) FormatError(xerrors.Printer) (next error) { return evt.error }
func (evt SectorSealFailed) apply(*SectorInfo) {}
func (evt SectorSealPreCommitFailed) FormatError(xerrors.Printer) (next error) { return evt.error }
func (evt SectorSealPreCommitFailed) apply(si *SectorInfo) {
si.InvalidProofs = 0 // reset counter
}
type SectorPreCommitFailed struct{ error }
type SectorChainPreCommitFailed struct{ error }
func (evt SectorPreCommitFailed) FormatError(xerrors.Printer) (next error) { return evt.error }
func (evt SectorPreCommitFailed) apply(*SectorInfo) {}
func (evt SectorChainPreCommitFailed) FormatError(xerrors.Printer) (next error) { return evt.error }
func (evt SectorChainPreCommitFailed) apply(*SectorInfo) {}
type SectorPreCommitted struct {
Message cid.Cid
@ -157,6 +168,16 @@ type SectorRetryWaitSeed struct{}
func (evt SectorRetryWaitSeed) apply(state *SectorInfo) {}
type SectorRetryComputeProof struct{}
func (evt SectorRetryComputeProof) apply(state *SectorInfo) {}
type SectorRetryInvalidProof struct{}
func (evt SectorRetryInvalidProof) apply(state *SectorInfo) {
state.InvalidProofs++
}
// Faults
type SectorFaulty struct{}

16
storage/sealing/fsm_test.go
View File

@ -36,9 +36,12 @@ func TestHappyPath(t *testing.T) {
}
m.planSingle(SectorPacked{})
require.Equal(m.t, m.state.State, api.Unsealed)
require.Equal(m.t, m.state.State, api.PreCommit1)
m.planSingle(SectorSealed{})
m.planSingle(SectorPreCommit1{})
require.Equal(m.t, m.state.State, api.PreCommit2)
m.planSingle(SectorPreCommit2{})
require.Equal(m.t, m.state.State, api.PreCommitting)
m.planSingle(SectorPreCommitted{})
@ -65,9 +68,12 @@ func TestSeedRevert(t *testing.T) {
}
m.planSingle(SectorPacked{})
require.Equal(m.t, m.state.State, api.Unsealed)
require.Equal(m.t, m.state.State, api.PreCommit1)
m.planSingle(SectorSealed{})
m.planSingle(SectorPreCommit1{})
require.Equal(m.t, m.state.State, api.PreCommit2)
m.planSingle(SectorPreCommit2{})
require.Equal(m.t, m.state.State, api.PreCommitting)
m.planSingle(SectorPreCommitted{})
@ -102,5 +108,5 @@ func TestPlanCommittingHandlesSectorCommitFailed(t *testing.T) {
require.NoError(t, planCommitting(events, m.state))
require.Equal(t, api.SectorStates[api.CommitFailed], api.SectorStates[m.state.State])
require.Equal(t, api.CommitFailed, m.state.State)
}

9
storage/sealing/sealing.go
View File

@ -43,6 +43,7 @@ type sealingApi interface { // TODO: trim down
StateMinerSectors(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error)
StateMinerProvingSet(context.Context, address.Address, types.TipSetKey) ([]*api.ChainSectorInfo, error)
StateMinerSectorSize(context.Context, address.Address, types.TipSetKey) (abi.SectorSize, error)
StateSectorPreCommitInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error)
StateWaitMsg(context.Context, cid.Cid) (*api.MsgLookup, error) // TODO: removeme eventually
StateGetActor(ctx context.Context, actor address.Address, ts types.TipSetKey) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
@ -72,11 +73,12 @@ type Sealing struct {
sealer sectorstorage.SectorManager
sectors *statemachine.StateGroup
tktFn TicketFn
sc SectorIDCounter
verif ffiwrapper.Verifier
tktFn TicketFn
}
func New(api sealingApi, events *events.Events, maddr address.Address, worker address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, tktFn TicketFn) *Sealing {
func New(api sealingApi, events *events.Events, maddr address.Address, worker address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, tktFn TicketFn) *Sealing {
s := &Sealing{
api: api,
events: events,
@ -84,8 +86,9 @@ func New(api sealingApi, events *events.Events, maddr address.Address, worker ad
maddr: maddr,
worker: worker,
sealer: sealer,
tktFn: tktFn,
sc: sc,
verif: verif,
tktFn: tktFn,
}
s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{})

53
storage/sealing/states.go
View File

@ -49,11 +49,11 @@ func (m *Sealing) handlePacking(ctx statemachine.Context, sector SectorInfo) err
return ctx.Send(SectorPacked{Pieces: pieces})
}
func (m *Sealing) handleUnsealed(ctx statemachine.Context, sector SectorInfo) error {
func (m *Sealing) handlePreCommit1(ctx statemachine.Context, sector SectorInfo) error {
if err := checkPieces(ctx.Context(), sector, m.api); err != nil { // Sanity check state
switch err.(type) {
case *ErrApi:
log.Errorf("handleUnsealed: api error, not proceeding: %+v", err)
log.Errorf("handlePreCommit1: api error, not proceeding: %+v", err)
return nil
case *ErrInvalidDeals:
return ctx.Send(SectorPackingFailed{xerrors.Errorf("invalid deals in sector: %w", err)})
@ -67,38 +67,44 @@ func (m *Sealing) handleUnsealed(ctx statemachine.Context, sector SectorInfo) er
log.Infow("performing sector replication...", "sector", sector.SectorID)
ticket, err := m.tktFn(ctx.Context())
if err != nil {
return ctx.Send(SectorSealFailed{xerrors.Errorf("getting ticket failed: %w", err)})
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("getting ticket failed: %w", err)})
}
pc1o, err := m.sealer.SealPreCommit1(ctx.Context(), m.minerSector(sector.SectorID), ticket.Value, sector.pieceInfos())
if err != nil {
return ctx.Send(SectorSealFailed{xerrors.Errorf("seal pre commit(1) failed: %w", err)})
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("seal pre commit(1) failed: %w", err)})
}
cids, err := m.sealer.SealPreCommit2(ctx.Context(), m.minerSector(sector.SectorID), pc1o)
return ctx.Send(SectorPreCommit1{
PreCommit1Out: pc1o,
Ticket: *ticket,
})
}
func (m *Sealing) handlePreCommit2(ctx statemachine.Context, sector SectorInfo) error {
cids, err := m.sealer.SealPreCommit2(ctx.Context(), m.minerSector(sector.SectorID), sector.PreCommit1Out)
if err != nil {
return ctx.Send(SectorSealFailed{xerrors.Errorf("seal pre commit(2) failed: %w", err)})
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("seal pre commit(2) failed: %w", err)})
}
return ctx.Send(SectorSealed{
return ctx.Send(SectorPreCommit2{
Unsealed: cids.Unsealed,
Sealed: cids.Sealed,
Ticket: *ticket,
})
}
func (m *Sealing) handlePreCommitting(ctx statemachine.Context, sector SectorInfo) error {
if err := checkSeal(ctx.Context(), m.maddr, sector, m.api); err != nil {
if err := checkPrecommit(ctx.Context(), m.maddr, sector, m.api); err != nil {
switch err.(type) {
case *ErrApi:
log.Errorf("handlePreCommitting: api error, not proceeding: %+v", err)
return nil
case *ErrBadCommD: // TODO: Should this just back to packing? (not really needed since handleUnsealed will do that too)
return ctx.Send(SectorSealFailed{xerrors.Errorf("bad CommD error: %w", err)})
case *ErrBadCommD: // TODO: Should this just back to packing? (not really needed since handlePreCommit1 will do that too)
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("bad CommD error: %w", err)})
case *ErrExpiredTicket:
return ctx.Send(SectorSealFailed{xerrors.Errorf("ticket expired: %w", err)})
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("ticket expired: %w", err)})
default:
return xerrors.Errorf("checkSeal sanity check error: %w", err)
return xerrors.Errorf("checkPrecommit sanity check error: %w", err)
}
}
@ -113,7 +119,7 @@ func (m *Sealing) handlePreCommitting(ctx statemachine.Context, sector SectorInf
}
enc, aerr := actors.SerializeParams(params)
if aerr != nil {
return ctx.Send(SectorPreCommitFailed{xerrors.Errorf("could not serialize commit sector parameters: %w", aerr)})
return ctx.Send(SectorChainPreCommitFailed{xerrors.Errorf("could not serialize commit sector parameters: %w", aerr)})
}
msg := &types.Message{
@ -129,7 +135,7 @@ func (m *Sealing) handlePreCommitting(ctx statemachine.Context, sector SectorInf
log.Info("submitting precommit for sector: ", sector.SectorID)
smsg, err := m.api.MpoolPushMessage(ctx.Context(), msg)
if err != nil {
return ctx.Send(SectorPreCommitFailed{xerrors.Errorf("pushing message to mpool: %w", err)})
return ctx.Send(SectorChainPreCommitFailed{xerrors.Errorf("pushing message to mpool: %w", err)})
}
return ctx.Send(SectorPreCommitted{Message: smsg.Cid()})
@ -140,17 +146,22 @@ func (m *Sealing) handleWaitSeed(ctx statemachine.Context, sector SectorInfo) er
log.Info("Sector precommitted: ", sector.SectorID)
mw, err := m.api.StateWaitMsg(ctx.Context(), *sector.PreCommitMessage)
if err != nil {
return ctx.Send(SectorPreCommitFailed{err})
return ctx.Send(SectorChainPreCommitFailed{err})
}
if mw.Receipt.ExitCode != 0 {
log.Error("sector precommit failed: ", mw.Receipt.ExitCode)
err := xerrors.Errorf("sector precommit failed: %d", mw.Receipt.ExitCode)
return ctx.Send(SectorPreCommitFailed{err})
return ctx.Send(SectorChainPreCommitFailed{err})
}
log.Info("precommit message landed on chain: ", sector.SectorID)
randHeight := mw.TipSet.Height() + miner.PreCommitChallengeDelay - 1 // -1 because of how the messages are applied
pci, err := m.api.StateSectorPreCommitInfo(ctx.Context(), m.maddr, sector.SectorID, mw.TipSet.Key())
if err != nil {
return xerrors.Errorf("getting precommit info: %w", err)
}
randHeight := pci.PreCommitEpoch + miner.PreCommitChallengeDelay
log.Infof("precommit for sector %d made it on chain, will start proof computation at height %d", sector.SectorID, randHeight)
err = m.events.ChainAt(func(ectx context.Context, ts *types.TipSet, curH abi.ChainEpoch) error {
@ -172,7 +183,7 @@ func (m *Sealing) handleWaitSeed(ctx statemachine.Context, sector SectorInfo) er
log.Warn("revert in interactive commit sector step")
// TODO: need to cancel running process and restart...
return nil
}, build.InteractivePoRepConfidence, mw.TipSet.Height()+miner.PreCommitChallengeDelay)
}, build.InteractivePoRepConfidence, randHeight)
if err != nil {
log.Warn("waitForPreCommitMessage ChainAt errored: ", err)
}
@ -199,6 +210,10 @@ func (m *Sealing) handleCommitting(ctx statemachine.Context, sector SectorInfo)
return ctx.Send(SectorComputeProofFailed{xerrors.Errorf("computing seal proof failed: %w", err)})
}
if err := m.checkCommit(ctx.Context(), sector, proof); err != nil {
return ctx.Send(SectorCommitFailed{xerrors.Errorf("commit check error: %w", err)})
}
// TODO: Consider splitting states and persist proof for faster recovery
params := &miner.ProveCommitSectorParams{

72
storage/sealing/states_failed.go
View File

@ -9,7 +9,6 @@ import (
"github.com/filecoin-project/specs-actors/actors/util/adt"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/apibstore"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@ -18,9 +17,11 @@ import (
const minRetryTime = 1 * time.Minute
func failedCooldown(ctx statemachine.Context, sector SectorInfo) error {
// TODO: Exponential backoff when we see consecutive failures
retryStart := time.Unix(int64(sector.Log[len(sector.Log)-1].Timestamp), 0).Add(minRetryTime)
if len(sector.Log) > 0 && !time.Now().After(retryStart) {
log.Infof("%s(%d), waiting %s before retrying", api.SectorStates[sector.State], sector.SectorID, time.Until(retryStart))
log.Infof("%s(%d), waiting %s before retrying", sector.State, sector.SectorID, time.Until(retryStart))
select {
case <-time.After(time.Until(retryStart)):
case <-ctx.Context().Done():
@ -75,17 +76,17 @@ func (m *Sealing) handleSealFailed(ctx statemachine.Context, sector SectorInfo)
}
func (m *Sealing) handlePreCommitFailed(ctx statemachine.Context, sector SectorInfo) error {
if err := checkSeal(ctx.Context(), m.maddr, sector, m.api); err != nil {
if err := checkPrecommit(ctx.Context(), m.maddr, sector, m.api); err != nil {
switch err.(type) {
case *ErrApi:
log.Errorf("handlePreCommitFailed: api error, not proceeding: %+v", err)
return nil
case *ErrBadCommD: // TODO: Should this just back to packing? (not really needed since handleUnsealed will do that too)
return ctx.Send(SectorSealFailed{xerrors.Errorf("bad CommD error: %w", err)})
case *ErrBadCommD: // TODO: Should this just back to packing? (not really needed since handlePreCommit1 will do that too)
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("bad CommD error: %w", err)})
case *ErrExpiredTicket:
return ctx.Send(SectorSealFailed{xerrors.Errorf("ticket expired error: %w", err)})
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("ticket expired error: %w", err)})
default:
return xerrors.Errorf("checkSeal sanity check error: %w", err)
return xerrors.Errorf("checkPrecommit sanity check error: %w", err)
}
}
@ -120,3 +121,60 @@ func (m *Sealing) handlePreCommitFailed(ctx statemachine.Context, sector SectorI
return ctx.Send(SectorRetryPreCommit{})
}
func (m *Sealing) handleComputeProofFailed(ctx statemachine.Context, sector SectorInfo) error {
// TODO: Check sector files
if err := failedCooldown(ctx, sector); err != nil {
return err
}
return ctx.Send(SectorRetryComputeProof{})
}
func (m *Sealing) handleCommitFailed(ctx statemachine.Context, sector SectorInfo) error {
if err := checkPrecommit(ctx.Context(), m.maddr, sector, m.api); err != nil {
switch err.(type) {
case *ErrApi:
log.Errorf("handleCommitFailed: api error, not proceeding: %+v", err)
return nil
case *ErrBadCommD:
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("bad CommD error: %w", err)})
case *ErrExpiredTicket:
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("ticket expired error: %w", err)})
default:
return xerrors.Errorf("checkPrecommit sanity check error: %w", err)
}
}
if err := m.checkCommit(ctx.Context(), sector, sector.Proof); err != nil {
switch err.(type) {
case *ErrApi:
log.Errorf("handleCommitFailed: api error, not proceeding: %+v", err)
return nil
case *ErrBadSeed:
log.Errorf("seed changed, will retry: %+v", err)
return ctx.Send(SectorRetryWaitSeed{})
case *ErrInvalidProof:
if err := failedCooldown(ctx, sector); err != nil {
return err
}
if sector.InvalidProofs > 0 {
return ctx.Send(SectorSealPreCommitFailed{xerrors.Errorf("consecutive invalid proofs")})
}
return ctx.Send(SectorRetryInvalidProof{})
default:
return xerrors.Errorf("checkCommit sanity check error: %w", err)
}
}
// TODO: Check sector files
if err := failedCooldown(ctx, sector); err != nil {
return err
}
return ctx.Send(SectorRetryComputeProof{})
}

15
storage/sealing/types.go
View File

@ -2,6 +2,7 @@ package sealing
import (
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/lotus/api"
@ -35,11 +36,14 @@ type SectorInfo struct {
Pieces []Piece
// PreCommit
CommD *cid.Cid
CommR *cid.Cid
Proof []byte
Ticket api.SealTicket
// PreCommit1
Ticket api.SealTicket
PreCommit1Out storage.PreCommit1Out
// PreCommit2
CommD *cid.Cid
CommR *cid.Cid
Proof []byte
PreCommitMessage *cid.Cid
@ -48,6 +52,7 @@ type SectorInfo struct {
// Committing
CommitMessage *cid.Cid
InvalidProofs uint64 // failed proof computations (doesn't validate with proof inputs)
// Faults
FaultReportMsg *cid.Cid

2
storage/sealing/types_test.go
View File

@ -18,7 +18,7 @@ func TestSectorInfoSelialization(t *testing.T) {
dummyCid := builtin.AccountActorCodeID
si := &SectorInfo{
State: 123,
State: "stateful",
SectorID: 234,
Nonce: 345,
Pieces: []Piece{{