Merge pull request #183 from filecoin-project/feat/fetch-v1.14.0-rc7

opt:  snapdeal

api/storage_struct.go

@@ -37,7 +37,6 @@ import (
 //  * Generate markdown docs
 //  * Generate openrpc blobs
 
-
 var ErrNotSupported = xerrors.New("method not supported")
 
 // StorageMiner is a low-level interface to the Filecoin network storage miner node
@@ -245,6 +244,7 @@ type StorageMinerStruct struct {
 		ReturnUnsealPiece               func(ctx context.Context, callID types.CallID, err *storiface.CallError) error                                             `perm:"admin" retry:"true"`
 		ReturnReadPiece                 func(ctx context.Context, callID types.CallID, ok bool, err *storiface.CallError) error                                    `perm:"admin" retry:"true"`
 		ReturnFetch                     func(ctx context.Context, callID types.CallID, err *storiface.CallError) error                                             `perm:"admin" retry:"true"`
+		ReturnFinalizeReplicaUpdate     func(ctx context.Context, callID types.CallID, err *storiface.CallError) error                                             `perm:"admin" retry:"true"`
 
 		SealingSchedDiag   func(context.Context, bool) (interface{}, error)   `perm:"admin"`
 		SealingAbort       func(ctx context.Context, call types.CallID) error `perm:"admin"`
@@ -292,9 +292,9 @@ type StorageMinerStruct struct {
 
 		CheckProvable func(ctx context.Context, pp abi.RegisteredPoStProof, sectors []storage.SectorRef, expensive bool) (map[abi.SectorNumber]string, error) `perm:"admin"`
 
-		MessagerWaitMessage func(ctx context.Context, uuid string, confidence uint64) (*types2.MsgLookup, error) `perm:"read"`
+		MessagerWaitMessage func(ctx context.Context, uuid string, confidence uint64) (*types2.MsgLookup, error)    `perm:"read"`
 		MessagerPushMessage func(ctx context.Context, msg *types2.Message, spec *messager.SendSpec) (string, error) `perm:"sign"`
-		MessagerGetMessage  func(ctx context.Context, uuid string) (*messager.Message, error)                      `perm:"write"`
+		MessagerGetMessage  func(ctx context.Context, uuid string) (*messager.Message, error)                       `perm:"write"`
 
 		IsUnsealed    func(ctx context.Context, sector storage.SectorRef, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize) (bool, error) `perm:"read"`
 		SectorsStatus func(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (SectorInfo, error)                                         `perm:"read"`
@@ -473,6 +473,13 @@ func (c *StorageMinerStruct) ReturnSealCommit2(ctx context.Context, callID types
 	return c.Internal.ReturnSealCommit2(ctx, callID, proof, err)
 }
 
+func (s *StorageMinerStruct) ReturnFinalizeReplicaUpdate(p0 context.Context, callID types.CallID, err *storiface.CallError) error {
+	if s.Internal.ReturnFinalizeReplicaUpdate == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.ReturnFinalizeReplicaUpdate(p0, callID, err)
+}
+
 func (c *StorageMinerStruct) ReturnFinalizeSector(ctx context.Context, callID types.CallID, err *storiface.CallError) error {
 	return c.Internal.ReturnFinalizeSector(ctx, callID, err)
 }
@@ -517,7 +524,6 @@ func (c *StorageMinerStruct) SealingSchedDiag(ctx context.Context, doSched bool)
 	return c.Internal.SealingSchedDiag(ctx, doSched)
 }
 
-
 func (s *StorageMinerStruct) SectorAbortUpgrade(p0 context.Context, p1 abi.SectorNumber) error {
 	if s.Internal.SectorAbortUpgrade == nil {
 		return ErrNotSupported

api/worker.go

@@ -31,6 +31,7 @@ type WorkerStruct struct {
 		SealPreCommit2            func(ctx context.Context, sector storage.SectorRef, pc1o storage.PreCommit1Out) (types.CallID, error)                                                                                     `perm:"admin"`
 		SealCommit1               func(ctx context.Context, sector storage.SectorRef, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, cids storage.SectorCids) (types.CallID, error) `perm:"admin"`
 		SealCommit2               func(ctx context.Context, sector storage.SectorRef, c1o storage.Commit1Out) (types.CallID, error)                                                                                         `perm:"admin"`
+		FinalizeReplicaUpdate     func(p0 context.Context, p1 storage.SectorRef, p2 []storage.Range) (types.CallID, error)                                                                                                  `perm:"admin"`
 		FinalizeSector            func(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) (types.CallID, error)                                                                                   `perm:"admin"`
 		ReplicaUpdate             func(ctx context.Context, sector storage.SectorRef, pieces []abi.PieceInfo) (types.CallID, error)                                                                                         `perm:"admin"`
 		ProveReplicaUpdate1       func(ctx context.Context, sector storage.SectorRef, sectorKey, newSealed, newUnsealed cid.Cid) (types.CallID, error)                                                                      `perm:"admin"`
@@ -98,6 +99,13 @@ func (w *WorkerStruct) SealCommit2(ctx context.Context, sector storage.SectorRef
 	return w.Internal.SealCommit2(ctx, sector, c1o)
 }
 
+func (s *WorkerStruct) FinalizeReplicaUpdate(p0 context.Context, p1 storage.SectorRef, p2 []storage.Range) (types.CallID, error) {
+	if s.Internal.FinalizeReplicaUpdate == nil {
+		return *new(types.CallID), ErrNotSupported
+	}
+	return s.Internal.FinalizeReplicaUpdate(p0, p1, p2)
+}
+
 func (w *WorkerStruct) FinalizeSector(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) (types.CallID, error) {
 	return w.Internal.FinalizeSector(ctx, sector, keepUnsealed)
 }

app/venus-sealer/util.go

@@ -85,6 +85,7 @@ var stateOrder = map[types2.SectorState]stateMeta{}
 var stateList = []stateMeta{
 	{col: 39, state: "Total"},
 	{col: color.FgGreen, state: types2.Proving},
+	{col: color.FgGreen, state: types2.UpdateActivating},
 
 	{col: color.FgBlue, state: types2.Empty},
 	{col: color.FgBlue, state: types2.WaitDeals},
@@ -115,6 +116,7 @@ var stateList = []stateMeta{
 	{col: color.FgYellow, state: types2.SubmitReplicaUpdate},
 	{col: color.FgYellow, state: types2.ReplicaUpdateWait},
 	{col: color.FgYellow, state: types2.FinalizeReplicaUpdate},
+	{col: color.FgYellow, state: types2.ReleaseSectorKey},
 
 	{col: color.FgCyan, state: types2.Terminating},
 	{col: color.FgCyan, state: types2.TerminateWait},
@@ -143,6 +145,7 @@ var stateList = []stateMeta{
 	{col: color.FgRed, state: types2.SnapDealsAddPieceFailed},
 	{col: color.FgRed, state: types2.SnapDealsDealsExpired},
 	{col: color.FgRed, state: types2.ReplicaUpdateFailed},
+	{col: color.FgRed, state: types2.ReleaseSectorKeyFailed},
 }
 
 func getActorAddress(ctx context.Context, nodeAPI api.StorageMiner, overrideMaddr string) (maddr address.Address, err error) {

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/filecoin-project/go-fil-markets v1.14.1
 	github.com/filecoin-project/go-jsonrpc v0.1.5
 	github.com/filecoin-project/go-padreader v0.0.1
-	github.com/filecoin-project/go-paramfetch v0.0.3-0.20220111000201-e42866db1a53
+	github.com/filecoin-project/go-paramfetch v0.0.4
 	github.com/filecoin-project/go-state-types v0.1.3
 	github.com/filecoin-project/go-statemachine v1.0.1
 	github.com/filecoin-project/go-statestore v0.2.0
@@ -35,7 +35,7 @@ require (
 	github.com/filecoin-project/specs-actors/v5 v5.0.4
 	github.com/filecoin-project/specs-actors/v6 v6.0.1
 	github.com/filecoin-project/specs-actors/v7 v7.0.0-rc1
-	github.com/filecoin-project/specs-storage v0.1.1-0.20211228030229-6d460d25a0c9
+	github.com/filecoin-project/specs-storage v0.2.0
 	github.com/filecoin-project/venus v1.2.0-rc5.0.20220210073318-71c52bbba9c5
 	github.com/filecoin-project/venus-market v1.0.2-0.20220210103815-5ea3e7f6c5ac
 	github.com/filecoin-project/venus-messager v1.4.0-rc2

go.sum

@@ -381,8 +381,9 @@ github.com/filecoin-project/go-padreader v0.0.0-20200903213702-ed5fae088b20/go.m
 github.com/filecoin-project/go-padreader v0.0.0-20210723183308-812a16dc01b1/go.mod h1:VYVPJqwpsfmtoHnAmPx6MUwmrK6HIcDqZJiuZhtmfLQ=
 github.com/filecoin-project/go-padreader v0.0.1 h1:8h2tVy5HpoNbr2gBRr+WD6zV6VD6XHig+ynSGJg8ZOs=
 github.com/filecoin-project/go-padreader v0.0.1/go.mod h1:VYVPJqwpsfmtoHnAmPx6MUwmrK6HIcDqZJiuZhtmfLQ=
-github.com/filecoin-project/go-paramfetch v0.0.3-0.20220111000201-e42866db1a53 h1:+nripp+UI/rhl01w9Gs4V0XDGaVPYPMGU/D/gNVLue0=
 github.com/filecoin-project/go-paramfetch v0.0.3-0.20220111000201-e42866db1a53/go.mod h1:1FH85P8U+DUEmWk1Jkw3Bw7FrwTVUNHk/95PSPG+dts=
+github.com/filecoin-project/go-paramfetch v0.0.4 h1:H+Me8EL8T5+79z/KHYQQcT8NVOzYVqXIi7nhb48tdm8=
+github.com/filecoin-project/go-paramfetch v0.0.4/go.mod h1:1FH85P8U+DUEmWk1Jkw3Bw7FrwTVUNHk/95PSPG+dts=
 github.com/filecoin-project/go-state-types v0.0.0-20200903145444-247639ffa6ad/go.mod h1:IQ0MBPnonv35CJHtWSN3YY1Hz2gkPru1Q9qoaYLxx9I=
 github.com/filecoin-project/go-state-types v0.0.0-20200928172055-2df22083d8ab/go.mod h1:ezYnPf0bNkTsDibL/psSz5dy4B5awOJ/E7P2Saeep8g=
 github.com/filecoin-project/go-state-types v0.0.0-20201102161440-c8033295a1fc/go.mod h1:ezYnPf0bNkTsDibL/psSz5dy4B5awOJ/E7P2Saeep8g=
@@ -415,8 +416,9 @@ github.com/filecoin-project/specs-actors/v7 v7.0.0-20211117170924-fd07a4c7dff9/g
 github.com/filecoin-project/specs-actors/v7 v7.0.0-20211222192039-c83bea50c402/go.mod h1:p6LIOFezA1rgRLMewbvdi3Pp6SAu+q9FtJ9CAleSjrE=
 github.com/filecoin-project/specs-actors/v7 v7.0.0-rc1 h1:FuDaXIbcw2hRsFI8SDTmsGGCE+NumpF6aiBoU/2X5W4=
 github.com/filecoin-project/specs-actors/v7 v7.0.0-rc1/go.mod h1:TA5FwCna+Yi36POaT7SLKXsgEDvJwc0V/L6ZsO19B9M=
-github.com/filecoin-project/specs-storage v0.1.1-0.20211228030229-6d460d25a0c9 h1:oUYOvF7EvdXS0Zmk9mNkaB6Bu0l+WXBYPzVodKMiLug=
 github.com/filecoin-project/specs-storage v0.1.1-0.20211228030229-6d460d25a0c9/go.mod h1:Tb88Zq+IBJbvAn3mS89GYj3jdRThBTE/771HCVZdRJU=
+github.com/filecoin-project/specs-storage v0.2.0 h1:Y4UDv0apRQ3zI2GiPPubi8JblpUZZphEdaJUxCutfyg=
+github.com/filecoin-project/specs-storage v0.2.0/go.mod h1:Tb88Zq+IBJbvAn3mS89GYj3jdRThBTE/771HCVZdRJU=
 github.com/filecoin-project/test-vectors/schema v0.0.5/go.mod h1:iQ9QXLpYWL3m7warwvK1JC/pTri8mnfEmKygNDqqY6E=
 github.com/filecoin-project/venus v1.2.0-rc5/go.mod h1:P2i2m4nheAeVXXWjkVLDyezY7l2Bi1sNBt5gno0/Sn0=
 github.com/filecoin-project/venus v1.2.0-rc5.0.20220210073318-71c52bbba9c5 h1:RnwvdAb3HTktA57dX8UCY45qcPiFY/2J+aXhqOJ42wY=

sector-storage/faults.go

@@ -55,6 +55,25 @@ func (m *Manager) CheckProvable(ctx context.Context, pp abi.RegisteredPoStProof,
 				return nil
 			}
 
+			// temporary hack to make the check work with snapdeals
+			// will go away in https://github.com/filecoin-project/lotus/pull/7971
+			if lp.Sealed == "" || lp.Cache == "" {
+				// maybe it's update
+				lockedUpdate, err := m.index.StorageTryLock(ctx, sector.ID, storiface.FTUpdate|storiface.FTUpdateCache, storiface.FTNone)
+				if err != nil {
+					return xerrors.Errorf("acquiring sector lock: %w", err)
+				}
+				if lockedUpdate {
+					lp, _, err = m.localStore.AcquireSector(ctx, sector, storiface.FTUpdate|storiface.FTUpdateCache, storiface.FTNone, storiface.PathStorage, storiface.AcquireMove)
+					if err != nil {
+						log.Warnw("CheckProvable Sector FAULT: acquire sector in checkProvable", "sector", sector, "error", err)
+						bad[sector.ID] = fmt.Sprintf("acquire sector failed: %s", err)
+						return nil
+					}
+					lp.Sealed, lp.Cache = lp.Update, lp.UpdateCache
+				}
+			}
+
 			if lp.Sealed == "" || lp.Cache == "" {
 				log.Warnw("CheckProvable Sector FAULT: cache and/or sealed paths not found", "sector", sector, "sealed", lp.Sealed, "cache", lp.Cache)
 				bad[sector.ID] = fmt.Sprintf("cache and/or sealed paths not found, cache %q, sealed %q", lp.Cache, lp.Sealed)

sector-storage/ffiwrapper/sealer_cgo.go

@@ -811,7 +811,7 @@ func (sb *Sealer) ReleaseSealed(ctx context.Context, sector storage.SectorRef) e
 	return xerrors.Errorf("not supported at this layer")
 }
 
-func (sb *Sealer) FinalizeSector(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
+func (sb *Sealer) freeUnsealed(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
 	ssize, err := sector.ProofType.SectorSize()
 	if err != nil {
 		return err
@@ -875,7 +875,18 @@ func (sb *Sealer) FinalizeSector(ctx context.Context, sector storage.SectorRef,
 		}
 
 	}
+	return nil
+}
+
+func (sb *Sealer) FinalizeSector(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
+	ssize, err := sector.ProofType.SectorSize()
+	if err != nil {
+		return err
+	}
 
+	if err := sb.freeUnsealed(ctx, sector, keepUnsealed); err != nil {
+		return err
+	}
 	paths, done, err := sb.sectors.AcquireSector(ctx, sector, storiface.FTCache, 0, storiface.PathStorage)
 	if err != nil {
 		return xerrors.Errorf("acquiring sector cache path: %w", err)
@@ -886,6 +897,43 @@ func (sb *Sealer) FinalizeSector(ctx context.Context, sector storage.SectorRef,
 	return ffi.ClearCache(uint64(ssize), paths.Cache)
 }
 
+func (sb *Sealer) FinalizeReplicaUpdate(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
+	ssize, err := sector.ProofType.SectorSize()
+	if err != nil {
+		return err
+	}
+
+	if err := sb.freeUnsealed(ctx, sector, keepUnsealed); err != nil {
+		return err
+	}
+
+	{
+		paths, done, err := sb.sectors.AcquireSector(ctx, sector, storiface.FTCache, 0, storiface.PathStorage)
+		if err != nil {
+			return xerrors.Errorf("acquiring sector cache path: %w", err)
+		}
+		defer done()
+
+		if err := ffi.ClearCache(uint64(ssize), paths.Cache); err != nil {
+			return xerrors.Errorf("clear cache: %w", err)
+		}
+	}
+
+	{
+		paths, done, err := sb.sectors.AcquireSector(ctx, sector, storiface.FTUpdateCache, 0, storiface.PathStorage)
+		if err != nil {
+			return xerrors.Errorf("acquiring sector cache path: %w", err)
+		}
+		defer done()
+
+		if err := ffi.ClearCache(uint64(ssize), paths.UpdateCache); err != nil {
+			return xerrors.Errorf("clear cache: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (sb *Sealer) ReleaseUnsealed(ctx context.Context, sector storage.SectorRef, safeToFree []storage.Range) error {
 	// This call is meant to mark storage as 'freeable'. Given that unsealing is
 	// very expensive, we don't remove data as soon as we can - instead we only

sector-storage/manager.go

@@ -150,8 +150,9 @@ func New(ctx context.Context, lstor *stores.Local, stor *stores.Remote, ls store
 	go m.sched.runSched()
 
 	localTasks := []types.TaskType{
-		types.TTCommit1, types.TTProveReplicaUpdate1, types.TTFinalize, types.TTFetch,
+		types.TTCommit1, types.TTProveReplicaUpdate1, types.TTFinalize, types.TTFetch, types.TTFinalizeReplicaUpdate,
 	}
+
 	if sc.AllowAddPiece {
 		localTasks = append(localTasks, types.TTAddPiece)
 	}
@@ -591,6 +592,74 @@ func (m *Manager) FinalizeSector(ctx context.Context, sector storage.SectorRef,
 	return nil
 }
 
+func (m *Manager) FinalizeReplicaUpdate(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	if err := m.index.StorageLock(ctx, sector.ID, storiface.FTNone, storiface.FTSealed|storiface.FTUnsealed|storiface.FTCache|storiface.FTUpdate|storiface.FTUpdateCache); err != nil {
+		return xerrors.Errorf("acquiring sector lock: %w", err)
+	}
+
+	fts := storiface.FTUnsealed
+	{
+		unsealedStores, err := m.index.StorageFindSector(ctx, sector.ID, storiface.FTUnsealed, 0, false)
+		if err != nil {
+			return xerrors.Errorf("finding unsealed sector: %w", err)
+		}
+
+		if len(unsealedStores) == 0 { // Is some edge-cases unsealed sector may not exist already, that's fine
+			fts = storiface.FTNone
+		}
+	}
+
+	pathType := storiface.PathStorage
+	{
+		sealedStores, err := m.index.StorageFindSector(ctx, sector.ID, storiface.FTUpdate, 0, false)
+		if err != nil {
+			return xerrors.Errorf("finding sealed sector: %w", err)
+		}
+
+		for _, store := range sealedStores {
+			if store.CanSeal {
+				pathType = storiface.PathSealing
+				break
+			}
+		}
+	}
+
+	selector := newExistingSelector(m.index, sector.ID, storiface.FTCache|storiface.FTSealed|storiface.FTUpdate|storiface.FTUpdateCache, false)
+
+	err := m.sched.Schedule(ctx, sector, types.TTFinalizeReplicaUpdate, selector,
+		m.schedFetch(sector, storiface.FTCache|storiface.FTSealed|storiface.FTUpdate|storiface.FTUpdateCache|fts, pathType, storiface.AcquireMove),
+		func(ctx context.Context, w Worker) error {
+			_, err := m.waitSimpleCall(ctx)(w.FinalizeReplicaUpdate(ctx, sector, keepUnsealed))
+			return err
+		})
+	if err != nil {
+		return err
+	}
+
+	fetchSel := newAllocSelector(m.index, storiface.FTCache|storiface.FTSealed|storiface.FTUpdate|storiface.FTUpdateCache, storiface.PathStorage)
+	moveUnsealed := fts
+	{
+		if len(keepUnsealed) == 0 {
+			moveUnsealed = storiface.FTNone
+		}
+	}
+
+	err = m.sched.Schedule(ctx, sector, types.TTFetch, fetchSel,
+		m.schedFetch(sector, storiface.FTCache|storiface.FTSealed|storiface.FTUpdate|storiface.FTUpdateCache|moveUnsealed, storiface.PathStorage, storiface.AcquireMove),
+		func(ctx context.Context, w Worker) error {
+			_, err := m.waitSimpleCall(ctx)(w.MoveStorage(ctx, sector, storiface.FTCache|storiface.FTSealed|storiface.FTUpdate|storiface.FTUpdateCache|moveUnsealed))
+			return err
+		})
+	if err != nil {
+		return xerrors.Errorf("moving sector to storage: %w", err)
+	}
+
+	return nil
+}
+
 func (m *Manager) ReleaseUnsealed(ctx context.Context, sector storage.SectorRef, safeToFree []storage.Range) error {
 	return nil
 }
@@ -890,6 +959,10 @@ func (m *Manager) ReturnProveReplicaUpdate2(ctx context.Context, callID types.Ca
 	return m.returnResult(ctx, callID, proof, err)
 }
 
+func (m *Manager) ReturnFinalizeReplicaUpdate(ctx context.Context, callID types.CallID, err *storiface.CallError) error {
+	return m.returnResult(ctx, callID, nil, err)
+}
+
 func (m *Manager) ReturnGenerateSectorKeyFromData(ctx context.Context, callID types.CallID, err *storiface.CallError) error {
 	return m.returnResult(ctx, callID, nil, err)
 }

sector-storage/mock/mock.go

@@ -479,6 +479,10 @@ func (mgr *SectorMgr) FinalizeSector(context.Context, storage.SectorRef, []stora
 	return nil
 }
 
+func (mgr *SectorMgr) FinalizeReplicaUpdate(context.Context, storage.SectorRef, []storage.Range) error {
+	return nil
+}
+
 func (mgr *SectorMgr) ReleaseUnsealed(ctx context.Context, sector storage.SectorRef, safeToFree []storage.Range) error {
 	return nil
 }
@@ -579,6 +583,10 @@ func (mgr *SectorMgr) ReturnGenerateSectorKeyFromData(ctx context.Context, callI
 	panic("not supported")
 }
 
+func (mgr *SectorMgr) ReturnFinalizeReplicaUpdate(ctx context.Context, callID types.CallID, err *storiface.CallError) error {
+	panic("not supported")
+}
+
 func (m mockVerifProver) VerifySeal(svi proof.SealVerifyInfo) (bool, error) {
 	plen, err := svi.SealProof.ProofSize()
 	if err != nil {

sector-storage/sched_test.go

@@ -94,6 +94,9 @@ func (s *schedTestWorker) Remove(ctx context.Context, sector storage.SectorRef)
 	panic("implement me")
 }
 
+func (t *testExec) FinalizeReplicaUpdate(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) error {
+	panic("implement me")
+}
 func (s *schedTestWorker) NewSector(ctx context.Context, sector storage.SectorRef) (types.CallID, error) {
 	panic("implement me")
 }
@@ -118,6 +121,10 @@ func (s *schedTestWorker) GenerateSectorKeyFromData(ctx context.Context, sector
 	panic("implement me")
 }
 
+func (s *schedTestWorker) FinalizeReplicaUpdate(ctx context.Context, sector storage.SectorRef, keepUnsealed []storage.Range) (types.CallID, error) {
+	panic("implement me")
+}
+
 func (s *schedTestWorker) MoveStorage(ctx context.Context, sector storage.SectorRef, types storiface.SectorFileType) (types.CallID, error) {
 	panic("implement me")
 }