diff --git a/drivers/google_drive/driver.go b/drivers/google_drive/driver.go index 94ef854f2f..305f9fb08d 100644 --- a/drivers/google_drive/driver.go +++ b/drivers/google_drive/driver.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "strconv" + "time" "github.com/OpenListTeam/OpenList/v4/drivers/base" "github.com/OpenListTeam/OpenList/v4/internal/driver" @@ -111,6 +112,40 @@ func (d *GoogleDrive) Remove(ctx context.Context, obj model.Obj) error { return err } +var replaceCleanupDelay = 30 * time.Second + +// Replace installs src at dst without invalidating readers that already resolved +// the previous destination object. Google Drive permits duplicate names, so the +// source is published first, then the previous destination is renamed out of the +// canonical namespace. Its file ID is kept alive briefly so in-flight readers can +// finish resolving/downloading it before asynchronous cleanup deletes it. +func (d *GoogleDrive) Replace(ctx context.Context, srcObj, dstObj model.Obj, dstName string) error { + publish := base.Json{"name": dstName} + url := "https://www.googleapis.com/drive/v3/files/" + srcObj.GetID() + _, err := d.request(url, http.MethodPatch, func(req *resty.Request) { + req.SetBody(publish) + }, nil) + if err != nil { + return err + } + + tombstoneName := ".openlist-replaced-" + dstObj.GetID() + tombstone := base.Json{"name": tombstoneName} + url = "https://www.googleapis.com/drive/v3/files/" + dstObj.GetID() + _, err = d.request(url, http.MethodPatch, func(req *resty.Request) { + req.SetBody(tombstone) + }, nil) + if err != nil { + return err + } + + time.AfterFunc(replaceCleanupDelay, func() { + url := "https://www.googleapis.com/drive/v3/files/" + dstObj.GetID() + _, _ = d.request(url, http.MethodDelete, nil, nil) + }) + return nil +} + func (d *GoogleDrive) Put(ctx context.Context, dstDir model.Obj, stream model.FileStreamer, up driver.UpdateProgress) error { obj := stream.GetExist() var ( diff --git a/drivers/google_drive/replace_test.go b/drivers/google_drive/replace_test.go new file mode 100644 index 0000000000..c45b32297f --- /dev/null +++ b/drivers/google_drive/replace_test.go @@ -0,0 +1,84 @@ +package google_drive + +import ( + "context" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/OpenListTeam/OpenList/v4/drivers/base" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/go-resty/resty/v2" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestReplacePublishesSourceBeforeRetiringDestination(t *testing.T) { + oldClient := base.RestyClient + oldDelay := replaceCleanupDelay + replaceCleanupDelay = 10 * time.Millisecond + client := resty.New() + base.RestyClient = client + t.Cleanup(func() { + base.RestyClient = oldClient + replaceCleanupDelay = oldDelay + }) + + var mu sync.Mutex + var requests []string + deleted := make(chan struct{}, 1) + client.SetTransport(roundTripFunc(func(r *http.Request) (*http.Response, error) { + body, _ := io.ReadAll(r.Body) + mu.Lock() + requests = append(requests, r.Method+" "+r.URL.Path+" "+string(body)) + mu.Unlock() + if r.Method == http.MethodDelete { + select { + case deleted <- struct{}{}: + default: + } + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{}`)), + Request: r, + }, nil + })) + + d := &GoogleDrive{} + src := &model.Object{ID: "source-id", Name: "staged"} + dst := &model.Object{ID: "destination-id", Name: "canonical"} + if err := d.Replace(context.Background(), src, dst, "canonical"); err != nil { + t.Fatalf("Replace() error = %v", err) + } + + mu.Lock() + immediate := append([]string(nil), requests...) + mu.Unlock() + if len(immediate) != 2 { + t.Fatalf("got %d immediate requests, want 2: %v", len(immediate), immediate) + } + if got := immediate[0]; !strings.HasPrefix(got, "PATCH /drive/v3/files/source-id ") || !strings.Contains(got, `"name":"canonical"`) { + t.Fatalf("first request must publish source under canonical name, got %q", got) + } + if got := immediate[1]; !strings.HasPrefix(got, "PATCH /drive/v3/files/destination-id ") || !strings.Contains(got, `"name":".openlist-replaced-destination-id"`) { + t.Fatalf("second request must retire old destination under tombstone name, got %q", got) + } + + select { + case <-deleted: + case <-time.After(time.Second): + t.Fatal("timed out waiting for asynchronous destination cleanup") + } + mu.Lock() + defer mu.Unlock() + if got := requests[len(requests)-1]; got != "DELETE /drive/v3/files/destination-id " { + t.Fatalf("final request must delete retired destination by id, got %q", got) + } +} diff --git a/internal/driver/driver.go b/internal/driver/driver.go index 373bb56534..6f3a5391cd 100644 --- a/internal/driver/driver.go +++ b/internal/driver/driver.go @@ -76,6 +76,12 @@ type Remove interface { Remove(ctx context.Context, obj model.Obj) error } +// Replace swaps src into dst while keeping dst continuously addressable. +// Drivers that can provide this stronger overwrite primitive should implement it. +type Replace interface { + Replace(ctx context.Context, srcObj, dstObj model.Obj, dstName string) error +} + type Put interface { // Put a file (provided as a FileStreamer) into the driver // Besides the most basic upload functionality, the following features also need to be implemented: diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 67a1ac065e..f0d1bc3f4f 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -100,6 +100,14 @@ func Rename(ctx context.Context, srcPath, dstName string, skipHook ...bool) erro return err } +func Replace(ctx context.Context, srcPath, dstPath string) error { + err := replace(ctx, srcPath, dstPath) + if err != nil { + log.Errorf("failed replace %s with %s: %+v", dstPath, srcPath, err) + } + return err +} + func Remove(ctx context.Context, path string) error { err := remove(ctx, path) if err != nil { diff --git a/internal/fs/other.go b/internal/fs/other.go index a23beb73bc..4ffa5d245e 100644 --- a/internal/fs/other.go +++ b/internal/fs/other.go @@ -5,6 +5,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/internal/conf" "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/errs" "github.com/OpenListTeam/OpenList/v4/internal/model" "github.com/OpenListTeam/OpenList/v4/internal/op" "github.com/OpenListTeam/OpenList/v4/internal/task" @@ -31,6 +32,21 @@ func rename(ctx context.Context, srcPath, dstName string, skipHook ...bool) erro return op.Rename(ctx, storage, srcActualPath, dstName) } +func replace(ctx context.Context, srcPath, dstPath string) error { + srcStorage, srcActualPath, err := op.GetStorageAndActualPath(srcPath) + if err != nil { + return errors.WithMessage(err, "failed get src storage") + } + dstStorage, dstActualPath, err := op.GetStorageAndActualPath(dstPath) + if err != nil { + return errors.WithMessage(err, "failed get dst storage") + } + if srcStorage.GetStorage().ID != dstStorage.GetStorage().ID { + return errors.WithStack(errs.NotSupport) + } + return op.Replace(ctx, srcStorage, srcActualPath, dstActualPath) +} + func remove(ctx context.Context, path string) error { storage, actualPath, err := op.GetStorageAndActualPath(path) if err != nil { diff --git a/internal/op/fs.go b/internal/op/fs.go index f82a3ca8f8..ccecf02ca6 100644 --- a/internal/op/fs.go +++ b/internal/op/fs.go @@ -445,6 +445,74 @@ func Move(ctx context.Context, storage driver.Driver, srcPath, dstDirPath string return nil } +// Replace atomically publishes src at dst for drivers that can keep dst continuously addressable. +// src and dst must be siblings; cross-directory replacement should use the normal move path. +func Replace(ctx context.Context, storage driver.Driver, srcPath, dstPath string) error { + if storage.Config().CheckStatus && storage.GetStorage().Status != WORK { + return errors.WithMessagef(errs.StorageNotInit, "storage status: %s", storage.GetStorage().Status) + } + srcPath = utils.FixAndCleanPath(srcPath) + dstPath = utils.FixAndCleanPath(dstPath) + if stdpath.Dir(srcPath) != stdpath.Dir(dstPath) { + return errors.WithStack(errs.NotSupport) + } + if utils.PathEqual(srcPath, "/") || utils.PathEqual(dstPath, "/") { + return errors.New("replace root folder is not allowed") + } + + srcRawObj, err := Get(ctx, storage, srcPath, true) + if err != nil { + return errors.WithMessage(err, "failed to get src object") + } + dstRawObj, err := Get(ctx, storage, dstPath, true) + if err != nil { + return errors.WithMessage(err, "failed to get dst object") + } + if model.ObjHasMask(srcRawObj, model.NoMove|model.NoRename) || model.ObjHasMask(dstRawObj, model.NoRemove) { + return errors.WithStack(errs.PermissionDenied) + } + + replacer, ok := storage.(driver.Replace) + if !ok { + return errors.WithStack(errs.NotImplement) + } + dstName := stdpath.Base(dstPath) + dirKey := Key(storage, stdpath.Dir(srcPath)) + var cachedDir *directoryCache + if !srcRawObj.IsDir() { + Cache.linkCache.DeleteKey(stdpath.Join(dirKey, srcRawObj.GetName())) + Cache.linkCache.DeleteKey(stdpath.Join(dirKey, dstRawObj.GetName())) + } + if !storage.Config().NoCache { + if cache, exist := Cache.dirCache.Get(dirKey); exist { + cachedDir = cache + // Publish src under the canonical cache key before the backend handoff. + // This prevents readers that already have a warm directory cache from + // following the soon-to-be-deleted destination object during Replace. + newObj := &model.ObjWrapName{Name: dstName, Obj: model.UnwrapObjName(srcRawObj)} + cache.UpdateObject(dstRawObj.GetName(), wrapObjName(storage, newObj)) + } + } + + if err := replacer.Replace(ctx, model.UnwrapObjName(srcRawObj), model.UnwrapObjName(dstRawObj), dstName); err != nil { + if cachedDir != nil { + cachedDir.UpdateObject(dstName, dstRawObj) + } + return errors.WithStack(err) + } + + if cachedDir != nil { + if srcRawObj.IsDir() { + Cache.deleteDirectoryTree(stdpath.Join(dirKey, srcRawObj.GetName())) + } + cachedDir.RemoveObject(srcRawObj.GetName()) + } + if ctx.Value(conf.SkipHookKey) == nil && needHandleObjsUpdateHook() { + go objsUpdateHook(context.WithoutCancel(ctx), storage, stdpath.Dir(dstPath), srcRawObj.IsDir()) + } + return nil +} + func Rename(ctx context.Context, storage driver.Driver, srcPath, dstName string) error { if storage.Config().CheckStatus && storage.GetStorage().Status != WORK { return errors.WithMessagef(errs.StorageNotInit, "storage status: %s", storage.GetStorage().Status) diff --git a/server/webdav/file.go b/server/webdav/file.go index ea60997359..e21971a19b 100644 --- a/server/webdav/file.go +++ b/server/webdav/file.go @@ -55,7 +55,22 @@ func moveFiles(ctx context.Context, src, dst string, overwrite bool) (status int if !common.CanWrite(user, srcMeta, srcDir) || !common.CanWrite(user, dstMeta, dstDir) { return http.StatusForbidden, nil } - if srcDir == dstDir { + dstExisted := false + if _, getErr := fs.Get(ctx, dst, &fs.GetArgs{NoLog: true}); getErr == nil { + dstExisted = true + if !overwrite { + return http.StatusPreconditionFailed, nil + } + } else if !errs.IsObjectNotFound(getErr) { + return http.StatusInternalServerError, getErr + } + + if dstExisted && srcDir == dstDir { + err = fs.Replace(ctx, src, dst) + if errors.Is(errors.Cause(err), errs.NotImplement) || errors.Is(errors.Cause(err), errs.NotSupport) { + err = fs.Rename(ctx, src, dstName) + } + } else if srcDir == dstDir { err = fs.Rename(ctx, src, dstName) } else { _, err = fs.Move(context.WithValue(ctx, conf.NoTaskKey, struct{}{}), src, dstDir) @@ -69,7 +84,9 @@ func moveFiles(ctx context.Context, src, dst string, overwrite bool) (status int if err != nil { return http.StatusInternalServerError, err } - // TODO if there are no files copy, should return 204 + if dstExisted { + return http.StatusNoContent, nil + } return http.StatusCreated, nil }