From 75cdaca78a8a6fe05624a8d004f56ea6d7010dc6 Mon Sep 17 00:00:00 2001 From: Victor Yang Date: Tue, 15 Sep 2026 20:16:32 +0800 Subject: [PATCH 1/2] fix(webdav): replace Google Drive objects without read gap --- drivers/google_drive/driver.go | 18 ++++++++ drivers/google_drive/replace_test.go | 57 +++++++++++++++++++++++ internal/driver/driver.go | 6 +++ internal/fs/fs.go | 8 ++++ internal/fs/other.go | 16 +++++++ internal/op/fs.go | 68 ++++++++++++++++++++++++++++ server/webdav/file.go | 21 ++++++++- 7 files changed, 192 insertions(+), 2 deletions(-) create mode 100644 drivers/google_drive/replace_test.go diff --git a/drivers/google_drive/driver.go b/drivers/google_drive/driver.go index 94ef854f2f..99e817caa2 100644 --- a/drivers/google_drive/driver.go +++ b/drivers/google_drive/driver.go @@ -111,6 +111,24 @@ func (d *GoogleDrive) Remove(ctx context.Context, obj model.Obj) error { return err } +// Replace installs src at dst before deleting the previous destination object. +// Google Drive permits duplicate names, so readers can continue resolving dst +// throughout the handoff instead of observing a delete-then-move gap. +func (d *GoogleDrive) Replace(ctx context.Context, srcObj, dstObj model.Obj, dstName string) error { + data := 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(data) + }, nil) + if err != nil { + return err + } + + url = "https://www.googleapis.com/drive/v3/files/" + dstObj.GetID() + _, err = d.request(url, http.MethodDelete, nil, nil) + return err +} + 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..491d11d7c4 --- /dev/null +++ b/drivers/google_drive/replace_test.go @@ -0,0 +1,57 @@ +package google_drive + +import ( + "context" + "io" + "net/http" + "strings" + "sync" + "testing" + + "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 TestReplacePublishesSourceBeforeDeletingDestination(t *testing.T) { + oldClient := base.RestyClient + client := resty.New() + base.RestyClient = client + t.Cleanup(func() { base.RestyClient = oldClient }) + + var mu sync.Mutex + var requests []string + 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() + 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) + } + + if len(requests) != 2 { + t.Fatalf("got %d requests, want 2: %v", len(requests), requests) + } + if got := requests[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 := requests[1]; got != "DELETE /drive/v3/files/destination-id " { + t.Fatalf("second request must delete old 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 } From 4c08593020de839bd70e1477187335ef146ab016 Mon Sep 17 00:00:00 2001 From: Victor Yang Date: Tue, 15 Sep 2026 21:00:27 +0800 Subject: [PATCH 2/2] fix(google-drive): drain readers before overwrite cleanup --- drivers/google_drive/driver.go | 31 ++++++++++++++++----- drivers/google_drive/replace_test.go | 41 +++++++++++++++++++++++----- 2 files changed, 58 insertions(+), 14 deletions(-) diff --git a/drivers/google_drive/driver.go b/drivers/google_drive/driver.go index 99e817caa2..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,22 +112,38 @@ func (d *GoogleDrive) Remove(ctx context.Context, obj model.Obj) error { return err } -// Replace installs src at dst before deleting the previous destination object. -// Google Drive permits duplicate names, so readers can continue resolving dst -// throughout the handoff instead of observing a delete-then-move gap. +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 { - data := base.Json{"name": dstName} + 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(data) + 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.MethodDelete, nil, nil) - return err + _, 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 { diff --git a/drivers/google_drive/replace_test.go b/drivers/google_drive/replace_test.go index 491d11d7c4..c45b32297f 100644 --- a/drivers/google_drive/replace_test.go +++ b/drivers/google_drive/replace_test.go @@ -7,6 +7,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/OpenListTeam/OpenList/v4/drivers/base" "github.com/OpenListTeam/OpenList/v4/internal/model" @@ -17,19 +18,31 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } -func TestReplacePublishesSourceBeforeDeletingDestination(t *testing.T) { +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 }) + 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), @@ -45,13 +58,27 @@ func TestReplacePublishesSourceBeforeDeletingDestination(t *testing.T) { t.Fatalf("Replace() error = %v", err) } - if len(requests) != 2 { - t.Fatalf("got %d requests, want 2: %v", len(requests), requests) + 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 := requests[0]; !strings.HasPrefix(got, "PATCH /drive/v3/files/source-id ") || !strings.Contains(got, `"name":"canonical"`) { + 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 := requests[1]; got != "DELETE /drive/v3/files/destination-id " { - t.Fatalf("second request must delete old destination by id, 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) } }