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

sealing: Handlef for PreCommitFailed

Browse Source
This commit is contained in:
Łukasz Magiera 2020-01-23 18:34:04 +01:00
parent 7e43c40529
commit de52d3cadd
6 changed files with 90 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -151,7 +151,7 @@ var sectorsListCmd = &cli.Command{
return list[i] < list[j]
})
w := tabwriter.NewWriter(os.Stdout, 8, 4, 0, ' ', 0)
w := tabwriter.NewWriter(os.Stdout, 8, 4, 1, ' ', 0)
for _, s := range list {
st, err := nodeApi.SectorsStatus(ctx, s)

2
storage/miner.go
View File

@ -50,7 +50,6 @@ type storageMinerApi interface {
StateGetActor(ctx context.Context, actor address.Address, ts *types.TipSet) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, *types.TipSet) (*types.MessageReceipt, error)
StateMarketStorageDeal(context.Context, uint64, *types.TipSet) (*actors.OnChainDeal, error)
StateReadState(ctx context.Context, act *types.Actor, ts *types.TipSet) (*api.ActorState, error)
MpoolPushMessage(context.Context, *types.Message) (*types.SignedMessage, error)
@ -59,6 +58,7 @@ type storageMinerApi interface {
ChainGetRandomness(context.Context, types.TipSetKey, int64) ([]byte, error)
ChainGetTipSetByHeight(context.Context, uint64, *types.TipSet) (*types.TipSet, error)
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
WalletSign(context.Context, address.Address, []byte) (*types.Signature, error)
WalletBalance(context.Context, address.Address) (types.BigInt, error)

6
storage/sealing/fsm.go
View File

@ -60,6 +60,10 @@ var fsmPlanners = []func(events []statemachine.Event, state *SectorInfo) error{
api.SealFailed: planOne(
on(SectorRetrySeal{}, api.Unsealed),
),
api.PreCommitFailed: planOne(
on(SectorRetryPreCommit{}, api.PreCommitting),
on(SectorRetryWaitSeed{}, api.WaitSeed),
),
api.Faulty: planOne(
on(SectorFaultReported{}, api.FaultReported),
@ -153,7 +157,7 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
case api.SealFailed:
return m.handleSealFailed, nil
case api.PreCommitFailed:
log.Warnf("sector %d entered unimplemented state 'PreCommitFailed'", state.SectorID)
return m.handlePreCommitFailed, nil
case api.SealCommitFailed:
log.Warnf("sector %d entered unimplemented state 'SealCommitFailed'", state.SectorID)
case api.CommitFailed:

8
storage/sealing/fsm_events.go
View File

@ -126,6 +126,14 @@ type SectorRetrySeal struct{}
func (evt SectorRetrySeal) apply(state *SectorInfo) {}
type SectorRetryPreCommit struct{}
func (evt SectorRetryPreCommit) apply(state *SectorInfo) {}
type SectorRetryWaitSeed struct{}
func (evt SectorRetryWaitSeed) apply(state *SectorInfo) {}
// Faults
type SectorFaulty struct{}

2
storage/sealing/sealing.go
View File

@ -39,7 +39,6 @@ type sealingApi interface { // TODO: trim down
StateGetActor(ctx context.Context, actor address.Address, ts *types.TipSet) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, *types.TipSet) (*types.MessageReceipt, error)
StateMarketStorageDeal(context.Context, uint64, *types.TipSet) (*actors.OnChainDeal, error)
StateReadState(ctx context.Context, act *types.Actor, ts *types.TipSet) (*api.ActorState, error)
MpoolPushMessage(context.Context, *types.Message) (*types.SignedMessage, error)
@ -48,6 +47,7 @@ type sealingApi interface { // TODO: trim down
ChainGetRandomness(context.Context, types.TipSetKey, int64) ([]byte, error)
ChainGetTipSetByHeight(context.Context, uint64, *types.TipSet) (*types.TipSet, error)
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
WalletSign(context.Context, address.Address, []byte) (*types.Signature, error)
WalletBalance(context.Context, address.Address) (types.BigInt, error)

84
storage/sealing/states_failed.go
View File

@ -1,10 +1,14 @@
package sealing
import (
"bytes"
"fmt"
"time"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/lib/statemachine"
)
@ -24,28 +28,41 @@ func failedCooldown(ctx statemachine.Context, sector SectorInfo) error {
return nil
}
func (m *Sealing) handleSealFailed(ctx statemachine.Context, sector SectorInfo) error {
// TODO: <Remove this section after we can re-precommit>
func (m *Sealing) checkPreCommitted(ctx statemachine.Context, sector SectorInfo) (*actors.PreCommittedSector, bool) {
act, err := m.api.StateGetActor(ctx.Context(), m.maddr, nil)
if err != nil {
log.Errorf("handleSealFailed(%d): temp error: %+v", sector.SectorID, err)
return nil
return nil, true
}
st, err := m.api.StateReadState(ctx.Context(), act, nil)
st, err := m.api.ChainReadObj(ctx.Context(), act.Head)
if err != nil {
log.Errorf("handleSealFailed(%d): temp error: %+v", sector.SectorID, err)
return nil
return nil, true
}
_, found := st.State.(map[string]interface{})["PreCommittedSectors"].(map[string]interface{})[fmt.Sprint(sector.SectorID)]
var state actors.StorageMinerActorState
if err := state.UnmarshalCBOR(bytes.NewReader(st)); err != nil {
log.Errorf("handleSealFailed(%d): temp error: unmarshaling miner state: %+v", sector.SectorID, err)
return nil, true
}
pci, found := state.PreCommittedSectors[fmt.Sprint(sector.SectorID)]
if found {
// TODO: If not expired yet, we can just try reusing sealticket
log.Errorf("sector found in miner preseal array: %+v", sector.SectorID, err)
return nil
log.Errorf("sector %d found in miner preseal array: %+v", sector.SectorID, err)
return pci, true
}
return nil, false
}
func (m *Sealing) handleSealFailed(ctx statemachine.Context, sector SectorInfo) error {
if _, is := m.checkPreCommitted(ctx, sector); is {
// TODO: Remove this after we can re-precommit
return nil // noop, for now
}
// </>
if err := failedCooldown(ctx, sector); err != nil {
return err
@ -53,3 +70,50 @@ func (m *Sealing) handleSealFailed(ctx statemachine.Context, sector SectorInfo)
return ctx.Send(SectorRetrySeal{})
}
func (m *Sealing) handlePreCommitFailed(ctx statemachine.Context, sector SectorInfo) error {
if err := checkSeal(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 *ErrExpiredTicket:
return ctx.Send(SectorSealFailed{xerrors.Errorf("bad CommD error: %w", err)})
default:
return xerrors.Errorf("checkSeal sanity check error: %w", err)
}
}
if pci, is := m.checkPreCommitted(ctx, sector); is && pci != nil {
if sector.PreCommitMessage != nil {
log.Warn("sector %d is precommitted on chain, but we don't have precommit message", sector.SectorID)
return nil // TODO: SeedWait needs this currently
}
if string(pci.Info.CommR) != string(sector.CommR) {
log.Warn("sector %d is precommitted on chain, with different CommR: %x != %x", sector.SectorID, pci.Info.CommR, sector.CommR)
return nil // TODO: remove when the actor allows re-precommit
}
// TODO: we could compare more things, but I don't think we really need to
// CommR tells us that CommD (and CommPs), and the ticket are all matching
if err := failedCooldown(ctx, sector); err != nil {
return err
}
return ctx.Send(SectorRetryWaitSeed{})
}
if sector.PreCommitMessage != nil {
log.Warn("retrying precommit even though the message failed to apply")
}
if err := failedCooldown(ctx, sector); err != nil {
return err
}
return ctx.Send(SectorRetryPreCommit{})
}