lotus/extern/storage-sealing/states_proving.go

148 lines
4.8 KiB
Go

package sealing
import (
"time"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-state-types/exitcode"
"github.com/filecoin-project/go-statemachine"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors/policy"
)
func (m *Sealing) handleFaulty(ctx statemachine.Context, sector SectorInfo) error {
// TODO: noop because this is now handled by the PoSt scheduler. We can reuse
// this state for tracking faulty sectors, or remove it when that won't be
// a breaking change
return nil
}
func (m *Sealing) handleFaultReported(ctx statemachine.Context, sector SectorInfo) error {
if sector.FaultReportMsg == nil {
return xerrors.Errorf("entered fault reported state without a FaultReportMsg cid")
}
mw, err := m.api.StateWaitMsg(ctx.Context(), *sector.FaultReportMsg)
if err != nil {
return xerrors.Errorf("failed to wait for fault declaration: %w", err)
}
if mw.Receipt.ExitCode != 0 {
log.Errorf("UNHANDLED: declaring sector fault failed (exit=%d, msg=%s) (id: %d)", mw.Receipt.ExitCode, *sector.FaultReportMsg, sector.SectorNumber)
return xerrors.Errorf("UNHANDLED: submitting fault declaration failed (exit %d)", mw.Receipt.ExitCode)
}
return ctx.Send(SectorFaultedFinal{})
}
func (m *Sealing) handleTerminating(ctx statemachine.Context, sector SectorInfo) error {
// First step of sector termination
// * See if sector is live
// * If not, goto removing
// * Add to termination queue
// * Wait for message to land on-chain
// * Check for correct termination
// * wait for expiration (+winning lookback?)
si, err := m.api.StateSectorGetInfo(ctx.Context(), m.maddr, sector.SectorNumber, nil)
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("getting sector info: %w", err)})
}
if si == nil {
// either already terminated or not committed yet
pci, err := m.api.StateSectorPreCommitInfo(ctx.Context(), m.maddr, sector.SectorNumber, nil)
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("checking precommit presence: %w", err)})
}
if pci != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("sector was precommitted but not proven, remove instead of terminating")})
}
return ctx.Send(SectorRemove{})
}
termCid, terminated, err := m.terminator.AddTermination(ctx.Context(), m.minerSectorID(sector.SectorNumber))
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("queueing termination: %w", err)})
}
if terminated {
return ctx.Send(SectorTerminating{Message: nil})
}
return ctx.Send(SectorTerminating{Message: &termCid})
}
func (m *Sealing) handleTerminateWait(ctx statemachine.Context, sector SectorInfo) error {
if sector.TerminateMessage == nil {
return xerrors.New("entered TerminateWait with nil TerminateMessage")
}
mw, err := m.api.StateWaitMsg(ctx.Context(), *sector.TerminateMessage)
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("waiting for terminate message to land on chain: %w", err)})
}
if mw.Receipt.ExitCode != exitcode.Ok {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("terminate message failed to execute: exit %d: %w", mw.Receipt.ExitCode, err)})
}
return ctx.Send(SectorTerminated{TerminatedAt: mw.Height})
}
func (m *Sealing) handleTerminateFinality(ctx statemachine.Context, sector SectorInfo) error {
for {
tok, epoch, err := m.api.ChainHead(ctx.Context())
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("getting chain head: %w", err)})
}
nv, err := m.api.StateNetworkVersion(ctx.Context(), tok)
if err != nil {
return ctx.Send(SectorTerminateFailed{xerrors.Errorf("getting network version: %w", err)})
}
if epoch >= sector.TerminatedAt+policy.GetWinningPoStSectorSetLookback(nv) {
return ctx.Send(SectorRemove{})
}
toWait := time.Duration(epoch-sector.TerminatedAt+policy.GetWinningPoStSectorSetLookback(nv)) * time.Duration(build.BlockDelaySecs) * time.Second
select {
case <-time.After(toWait):
continue
case <-ctx.Context().Done():
return ctx.Context().Err()
}
}
}
func (m *Sealing) handleRemoving(ctx statemachine.Context, sector SectorInfo) error {
if err := m.sealer.Remove(ctx.Context(), m.minerSector(sector.SectorType, sector.SectorNumber)); err != nil {
return ctx.Send(SectorRemoveFailed{err})
}
return ctx.Send(SectorRemoved{})
}
func (m *Sealing) handleProvingSector(ctx statemachine.Context, sector SectorInfo) error {
// TODO: track sector health / expiration
log.Infof("Proving sector %d", sector.SectorNumber)
cfg, err := m.getConfig()
if err != nil {
return xerrors.Errorf("getting sealing config: %w", err)
}
if err := m.sealer.ReleaseUnsealed(ctx.Context(), m.minerSector(sector.SectorType, sector.SectorNumber), sector.keepUnsealedRanges(true, cfg.AlwaysKeepUnsealedCopy)); err != nil {
log.Error(err)
}
// TODO: Watch termination
// TODO: Auto-extend if set
return nil
}