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

fix: fsm: shutdown removed sectors FSMs

Browse Source
This commit is contained in:
Łukasz Magiera 2023-02-28 09:48:17 +01:00
parent 0c9f697bf6
commit 04fe9062a6
1 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
storage/pipeline/fsm.go
View File

@ -6,8 +6,10 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"net/http" "net/http"
"os"
"reflect" "reflect"
"time" "time"
@ -19,9 +21,15 @@ import (
"github.com/filecoin-project/lotus/api" "github.com/filecoin-project/lotus/api"
) )
var errSectorRemoved = errors.New("sector removed")
func (m *Sealing) Plan(events []statemachine.Event, user interface{}) (interface{}, uint64, error) { func (m *Sealing) Plan(events []statemachine.Event, user interface{}) (interface{}, uint64, error) {
next, processed, err := m.plan(events, user.(*SectorInfo)) next, processed, err := m.plan(events, user.(*SectorInfo))
if err != nil || next == nil { if err != nil || next == nil {
if err == errSectorRemoved && os.Getenv("LOTUS_KEEP_REMOVED_FSM_ACTIVE") != "1" {
return nil, processed, statemachine.ErrTerminated
}
l := Log{ l := Log{
Timestamp: uint64(time.Now().Unix()), Timestamp: uint64(time.Now().Unix()),
Message: fmt.Sprintf("state machine error: %s", err), Message: fmt.Sprintf("state machine error: %s", err),
@ -601,7 +609,7 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
case Removing: case Removing:
return m.handleRemoving, processed, nil return m.handleRemoving, processed, nil
case Removed: case Removed:
return nil, processed, nil return nil, processed, errSectorRemoved
case RemoveFailed: case RemoveFailed:
return m.handleRemoveFailed, processed, nil return m.handleRemoveFailed, processed, nil
@ -615,13 +623,14 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
// Fatal errors // Fatal errors
case UndefinedSectorState: case UndefinedSectorState:
log.Error("sector update with undefined state!") log.Error("sector update with undefined state!")
return nil, processed, xerrors.Errorf("sector update with undefined state")
case FailedUnrecoverable: case FailedUnrecoverable:
log.Errorf("sector %d failed unrecoverably", state.SectorNumber) log.Errorf("sector %d failed unrecoverably", state.SectorNumber)
return nil, processed, xerrors.Errorf("sector %d failed unrecoverably", state.SectorNumber)
default: default:
log.Errorf("unexpected sector update state: %s", state.State) log.Errorf("unexpected sector update state: %s", state.State)
return nil, processed, xerrors.Errorf("unexpected sector update state: %s", state.State)
} }
return nil, processed, nil
} }
func (m *Sealing) onUpdateSector(ctx context.Context, state *SectorInfo) error { func (m *Sealing) onUpdateSector(ctx context.Context, state *SectorInfo) error {