diff --git a/extern/storage-sealing/fsm_events.go b/extern/storage-sealing/fsm_events.go index 167217080..e28366721 100644 --- a/extern/storage-sealing/fsm_events.go +++ b/extern/storage-sealing/fsm_events.go @@ -323,10 +323,10 @@ func (evt SectorTerminate) applyGlobal(state *SectorInfo) bool { return true } -type SectorTerminating struct{ Message cid.Cid } +type SectorTerminating struct{ Message *cid.Cid } func (evt SectorTerminating) apply(state *SectorInfo) { - state.TerminateMessage = &evt.Message + state.TerminateMessage = evt.Message } type SectorTerminated struct{ TerminatedAt abi.ChainEpoch } diff --git a/extern/storage-sealing/sealing.go b/extern/storage-sealing/sealing.go index 891bf8a1a..96d63efdc 100644 --- a/extern/storage-sealing/sealing.go +++ b/extern/storage-sealing/sealing.go @@ -62,6 +62,7 @@ type SealingAPI interface { StateMarketStorageDeal(context.Context, abi.DealID, TipSetToken) (market.DealProposal, error) StateNetworkVersion(ctx context.Context, tok TipSetToken) (network.Version, error) StateMinerProvingDeadline(context.Context, address.Address, TipSetToken) (*dline.Info, error) + StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tok TipSetToken) ([]api.Partition, error) SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error) ChainGetRandomnessFromBeacon(ctx context.Context, tok TipSetToken, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) diff --git a/extern/storage-sealing/states_proving.go b/extern/storage-sealing/states_proving.go index d59db7e75..212fd906f 100644 --- a/extern/storage-sealing/states_proving.go +++ b/extern/storage-sealing/states_proving.go @@ -7,7 +7,6 @@ import ( "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" ) @@ -65,12 +64,16 @@ func (m *Sealing) handleTerminating(ctx statemachine.Context, sector SectorInfo) return ctx.Send(SectorRemove{}) } - termCid, err := m.terminator.AddTermination(ctx.Context(), m.minerSectorID(sector.SectorNumber)) + 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)}) } - return ctx.Send(SectorTerminating{Message: termCid}) + if terminated { + return ctx.Send(SectorTerminating{Message: nil}) + } + + return ctx.Send(SectorTerminating{Message: &termCid}) } func (m *Sealing) handleTerminateWait(ctx statemachine.Context, sector SectorInfo) error { diff --git a/extern/storage-sealing/terminate_batch.go b/extern/storage-sealing/terminate_batch.go index 2eae0b08f..b86782808 100644 --- a/extern/storage-sealing/terminate_batch.go +++ b/extern/storage-sealing/terminate_batch.go @@ -34,6 +34,7 @@ type TerminateBatcherApi interface { SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) StateMinerInfo(context.Context, address.Address, TipSetToken) (miner.MinerInfo, error) StateMinerProvingDeadline(context.Context, address.Address, TipSetToken) (*dline.Info, error) + StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tok TipSetToken) ([]api.Partition, error) } type TerminateBatcher struct { @@ -45,7 +46,7 @@ type TerminateBatcher struct { todo map[SectorLocation]*bitfield.BitField // MinerSectorLocation -> BitField - waiting map[SectorLocation][]chan cid.Cid + waiting map[abi.SectorNumber][]chan cid.Cid notify, stop, stopped chan struct{} force chan chan *cid.Cid @@ -61,7 +62,7 @@ func NewTerminationBatcher(mctx context.Context, maddr address.Address, api Term feeCfg: feeCfg, todo: map[SectorLocation]*bitfield.BitField{}, - waiting: map[SectorLocation][]chan cid.Cid{}, + waiting: map[abi.SectorNumber][]chan cid.Cid{}, notify: make(chan struct{}, 1), force: make(chan chan *cid.Cid), @@ -209,32 +210,53 @@ func (b *TerminateBatcher) processBatch(notif, after bool) (*cid.Cid, error) { Deadline: t.Deadline, Partition: t.Partition, }) - } - for _, w := range b.waiting { - for _, ch := range w { - ch <- mcid // buffered + err := t.Sectors.ForEach(func(sn uint64) error { + for _, ch := range b.waiting[abi.SectorNumber(sn)] { + ch <- mcid // buffered + } + delete(b.waiting, abi.SectorNumber(sn)) + + return nil + }) + if err != nil { + return nil, xerrors.Errorf("sectors foreach: %w", err) } } - b.waiting = map[SectorLocation][]chan cid.Cid{} - return &mcid, nil } // register termination, wait for batch message, return message CID -func (b *TerminateBatcher) AddTermination(ctx context.Context, s abi.SectorID) (cid.Cid, error) { +// can return cid.Undef,true if the sector is already terminated on-chain +func (b *TerminateBatcher) AddTermination(ctx context.Context, s abi.SectorID) (mcid cid.Cid, terminated bool, err error) { maddr, err := address.NewIDAddress(uint64(s.Miner)) if err != nil { - return cid.Undef, err + return cid.Undef, false, err } loc, err := b.api.StateSectorPartition(ctx, maddr, s.Number, nil) if err != nil { - return cid.Undef, xerrors.Errorf("getting sector location: %w", err) + return cid.Undef, false, xerrors.Errorf("getting sector location: %w", err) } if loc == nil { - return cid.Undef, xerrors.New("sector location not found") + return cid.Undef, false, xerrors.New("sector location not found") + } + + { + // check if maybe already terminated + parts, err := b.api.StateMinerPartitions(ctx, maddr, loc.Deadline, nil) + if err != nil { + return cid.Cid{}, false, xerrors.Errorf("getting partitions: %w", err) + } + live, err := parts[loc.Partition].LiveSectors.IsSet(uint64(s.Number)) + if err != nil { + return cid.Cid{}, false, xerrors.Errorf("checking if sector is in live set: %w", err) + } + if !live { + // already terminated + return cid.Undef, true, nil + } } b.lk.Lock() @@ -247,7 +269,7 @@ func (b *TerminateBatcher) AddTermination(ctx context.Context, s abi.SectorID) ( bf.Set(uint64(s.Number)) sent := make(chan cid.Cid, 1) - b.waiting[*loc] = append(b.waiting[*loc], sent) + b.waiting[s.Number] = append(b.waiting[s.Number], sent) select { case b.notify <- struct{}{}: @@ -257,9 +279,9 @@ func (b *TerminateBatcher) AddTermination(ctx context.Context, s abi.SectorID) ( select { case c := <-sent: - return c, nil + return c, false, nil case <-ctx.Done(): - return cid.Undef, ctx.Err() + return cid.Undef, false, ctx.Err() } } diff --git a/storage/adapter_storage_miner.go b/storage/adapter_storage_miner.go index 0aa1ad2c7..20bf30825 100644 --- a/storage/adapter_storage_miner.go +++ b/storage/adapter_storage_miner.go @@ -243,6 +243,15 @@ func (s SealingAPIAdapter) StateSectorPartition(ctx context.Context, maddr addre return nil, nil // not found } +func (s SealingAPIAdapter) StateMinerPartitions(ctx context.Context, maddr address.Address, dlIdx uint64, tok sealing.TipSetToken) ([]api.Partition, error) { + tsk, err := types.TipSetKeyFromBytes(tok) + if err != nil { + return nil, xerrors.Errorf("failed to unmarshal TipSetToken to TipSetKey: %w", err) + } + + return s.delegate.StateMinerPartitions(ctx, maddr, dlIdx, tsk) +} + func (s SealingAPIAdapter) StateMarketStorageDeal(ctx context.Context, dealID abi.DealID, tok sealing.TipSetToken) (market.DealProposal, error) { tsk, err := types.TipSetKeyFromBytes(tok) if err != nil {