Merge pull request #9153 from filecoin-project/sbansal/detach-storage-on-worker-shutdown

fix: worker: Close all storage paths on worker shutdown

api/api_worker.go

@@ -62,6 +62,7 @@ type Worker interface {
 	StorageLocal(ctx context.Context) (map[storiface.ID]string, error)                   //perm:admin
 	StorageAddLocal(ctx context.Context, path string) error                              //perm:admin
 	StorageDetachLocal(ctx context.Context, path string) error                           //perm:admin
+	StorageDetachAll(ctx context.Context) error                                          //perm:admin
 	StorageRedeclareLocal(ctx context.Context, id *storiface.ID, dropMissing bool) error //perm:admin
 
 	// SetEnabled marks the worker as enabled/disabled. Not that this setting

api/proxy_gen.go

@@ -981,6 +981,8 @@ type WorkerStruct struct {
 
 		StorageAddLocal func(p0 context.Context, p1 string) error `perm:"admin"`
 
+		StorageDetachAll func(p0 context.Context) error `perm:"admin"`
+
 		StorageDetachLocal func(p0 context.Context, p1 string) error `perm:"admin"`
 
 		StorageLocal func(p0 context.Context) (map[storiface.ID]string, error) `perm:"admin"`
@@ -5668,6 +5670,17 @@ func (s *WorkerStub) StorageAddLocal(p0 context.Context, p1 string) error {
 	return ErrNotSupported
 }
 
+func (s *WorkerStruct) StorageDetachAll(p0 context.Context) error {
+	if s.Internal.StorageDetachAll == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.StorageDetachAll(p0)
+}
+
+func (s *WorkerStub) StorageDetachAll(p0 context.Context) error {
+	return ErrNotSupported
+}
+
 func (s *WorkerStruct) StorageDetachLocal(p0 context.Context, p1 string) error {
 	if s.Internal.StorageDetachLocal == nil {
 		return ErrNotSupported

cmd/lotus-worker/main.go

@@ -129,6 +129,13 @@ var stopCmd = &cli.Command{
 		defer closer()
 
 		ctx := lcli.ReqContext(cctx)
+
+		// Detach any storage associated with this worker
+		err = api.StorageDetachAll(ctx)
+		if err != nil {
+			return err
+		}
+
 		err = api.Shutdown(ctx)
 		if err != nil {
 			return err

cmd/lotus-worker/sealworker/rpc.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
+	logging "github.com/ipfs/go-log/v2"
 	"github.com/mitchellh/go-homedir"
 	"golang.org/x/xerrors"
 
@@ -23,6 +24,8 @@ import (
 	"github.com/filecoin-project/lotus/storage/sealer/storiface"
 )
 
+var log = logging.Logger("sealworker")
+
 func WorkerHandler(authv func(ctx context.Context, token string) ([]auth.Permission, error), remote http.HandlerFunc, a api.Worker, permissioned bool) http.Handler {
 	mux := mux.NewRouter()
 	readerHandler, readerServerOpt := rpcenc.ReaderParamDecoder()
@@ -146,6 +149,23 @@ func (w *Worker) StorageDetachLocal(ctx context.Context, path string) error {
 	return w.LocalStore.ClosePath(ctx, localPath.ID)
 }
 
+func (w *Worker) StorageDetachAll(ctx context.Context) error {
+
+	lps, err := w.LocalStore.Local(ctx)
+	if err != nil {
+		return xerrors.Errorf("getting local path list: %w", err)
+	}
+
+	for _, lp := range lps {
+		err = w.LocalStore.ClosePath(ctx, lp.ID)
+		if err != nil {
+			log.Warnf("unable to close path: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (w *Worker) StorageRedeclareLocal(ctx context.Context, id *storiface.ID, dropMissing bool) error {
 	return w.LocalStore.Redeclare(ctx, id, dropMissing)
 }

documentation/en/api-v0-methods-worker.md

@@ -39,6 +39,7 @@
   * [SetEnabled](#SetEnabled)
 * [Storage](#Storage)
   * [StorageAddLocal](#StorageAddLocal)
+  * [StorageDetachAll](#StorageDetachAll)
   * [StorageDetachLocal](#StorageDetachLocal)
   * [StorageLocal](#StorageLocal)
   * [StorageRedeclareLocal](#StorageRedeclareLocal)
@@ -2121,6 +2122,15 @@ Inputs:
 
 Response: `{}`
 
+### StorageDetachAll
+
+
+Perms: admin
+
+Inputs: `null`
+
+Response: `{}`
+
 ### StorageDetachLocal