From d44d85dc05d3da9394d5a3f1c5db14ec2badf3ed Mon Sep 17 00:00:00 2001 From: Federico Justus Denkena Date: Wed, 9 Sep 2026 23:36:25 +0200 Subject: [PATCH] bisync: add Dropbox list_folder cursor listings via --delta-list Skip the per-directory Dropbox walk on successive bisync runs by applying list_folder/continue deltas to the stored listing. Cursor reset, --max-delete, and atomic listing+cursor writes abort without a half-applied snapshot. The local side still walks; there is no NTFS USN dependency. --- backend/dropbox/dropbox.go | 111 ++++++++++++++ backend/dropbox/dropbox_internal_test.go | 33 ++++ cmd/bisync/cmd.go | 2 + cmd/bisync/delta_list.go | 152 +++++++++++++++++++ cmd/bisync/delta_list_run.go | 182 +++++++++++++++++++++++ cmd/bisync/delta_list_test.go | 122 +++++++++++++++ cmd/bisync/march.go | 3 + cmd/bisync/rc.go | 3 + 8 files changed, 608 insertions(+) create mode 100644 cmd/bisync/delta_list.go create mode 100644 cmd/bisync/delta_list_run.go create mode 100644 cmd/bisync/delta_list_test.go diff --git a/backend/dropbox/dropbox.go b/backend/dropbox/dropbox.go index 8ee55eb1d..04b777181 100644 --- a/backend/dropbox/dropbox.go +++ b/backend/dropbox/dropbox.go @@ -2292,6 +2292,117 @@ func (o *Object) Remove(ctx context.Context) (err error) { return err } +// ListingChange is one list_folder/continue entry for bisync --delta-list. +type ListingChange struct { + Remote string + Deleted bool + IsDir bool + Size int64 + ModTime time.Time + Hash string +} + +func isListFolderReset(err error) bool { + var cont files.ListFolderContinueAPIError + if errors.As(err, &cont) && cont.EndpointError != nil && cont.EndpointError.Tag == files.ListFolderContinueErrorReset { + return true + } + return false +} + +// ListFolderDeltas returns Dropbox list_folder changes since cursor. +// An empty cursor performs a recursive snapshot (all entries as non-deletes). +// reset is true when Dropbox invalidated the cursor; deltas will be empty. +func (f *Fs) ListFolderDeltas(ctx context.Context, cursor string) (newCursor string, reset bool, deltas []ListingChange, err error) { + if f.opt.SharedFiles || f.opt.SharedFolders { + return "", false, nil, errors.New("dropbox: ListFolderDeltas is not supported with shared_files or shared_folders") + } + started := false + var res *files.ListFolderResult + for { + if !started { + if cursor == "" { + arg := files.NewListFolderArg(f.opt.Enc.FromStandardPath(f.slashRoot)) + arg.Recursive = true + arg.Limit = 1000 + if arg.Path == "/" { + arg.Path = "" + } + err = f.pacer.Call(func() (bool, error) { + res, err = f.srv.ListFolderContext(ctx, arg) + return shouldRetry(ctx, err) + }) + } else { + arg := files.ListFolderContinueArg{Cursor: cursor} + err = f.pacer.Call(func() (bool, error) { + res, err = f.srv.ListFolderContinueContext(ctx, &arg) + return shouldRetry(ctx, err) + }) + } + if err != nil { + if isListFolderReset(err) { + return "", true, nil, nil + } + return "", false, nil, err + } + started = true + } else { + arg := files.ListFolderContinueArg{Cursor: res.Cursor} + err = f.pacer.Call(func() (bool, error) { + res, err = f.srv.ListFolderContinueContext(ctx, &arg) + return shouldRetry(ctx, err) + }) + if err != nil { + if isListFolderReset(err) { + return "", true, nil, nil + } + return "", false, nil, fmt.Errorf("list continue: %w", err) + } + } + for _, entry := range res.Entries { + ch, ok := f.listingChangeFromEntry(entry) + if !ok { + continue + } + deltas = append(deltas, ch) + } + newCursor = res.Cursor + if !res.HasMore { + break + } + } + return newCursor, false, deltas, nil +} + +func (f *Fs) listingChangeFromEntry(entry files.IsMetadata) (ListingChange, bool) { + var ( + ch ListingChange + entryPath string + ) + switch info := entry.(type) { + case *files.FolderMetadata: + ch.IsDir = true + entryPath = info.PathDisplay + case *files.FileMetadata: + entryPath = info.PathDisplay + ch.Size = int64(info.Size) + ch.ModTime = time.Time(info.ClientModified) + ch.Hash = info.ContentHash + case *files.DeletedMetadata: + ch.Deleted = true + ch.IsDir = true + entryPath = info.PathDisplay + default: + return ListingChange{}, false + } + entryPath = trimPrefixFold(entryPath, f.slashRootSlash) + if entryPath == "" { + return ListingChange{}, false + } + ch.Remote = f.opt.Enc.ToStandardPath(entryPath) + return ch, true +} + // Check the interfaces are satisfied var ( _ fs.Fs = (*Fs)(nil) diff --git a/backend/dropbox/dropbox_internal_test.go b/backend/dropbox/dropbox_internal_test.go index 78a701f69..fb30352da 100644 --- a/backend/dropbox/dropbox_internal_test.go +++ b/backend/dropbox/dropbox_internal_test.go @@ -535,3 +535,36 @@ func TestInternalFindSharedFileCaseInsensitive(t *testing.T) { _, err := f.findSharedFile(context.Background(), "no-such-file.txt") assert.ErrorIs(t, err, fs.ErrorObjectNotFound) } + +func TestListingChangeFromEntry(t *testing.T) { + f := &Fs{ + slashRootSlash: "/Dropbox/", + opt: Options{Enc: encoder.Standard}, + } + file := &files.FileMetadata{ + Metadata: files.Metadata{ + Name: "a.txt", + PathDisplay: "/Dropbox/sub/a.txt", + }, + Size: 42, + ContentHash: "abc", + ClientModified: dropbox.DBXTime(time.Date(2024, 3, 1, 0, 0, 0, 0, time.UTC)), + } + ch, ok := f.listingChangeFromEntry(file) + require.True(t, ok) + assert.Equal(t, "sub/a.txt", ch.Remote) + assert.Equal(t, int64(42), ch.Size) + assert.Equal(t, "abc", ch.Hash) + assert.False(t, ch.Deleted) + + del := &files.DeletedMetadata{ + Metadata: files.Metadata{ + Name: "sub", + PathDisplay: "/Dropbox/sub", + }, + } + ch, ok = f.listingChangeFromEntry(del) + require.True(t, ok) + assert.Equal(t, "sub", ch.Remote) + assert.True(t, ch.Deleted) +} diff --git a/cmd/bisync/cmd.go b/cmd/bisync/cmd.go index f1b3367bb..6b7fc3c22 100644 --- a/cmd/bisync/cmd.go +++ b/cmd/bisync/cmd.go @@ -61,6 +61,7 @@ type Options struct { ConflictSuffixFlag string ConflictSuffix1 string ConflictSuffix2 string + DeltaList bool } // Default values @@ -151,6 +152,7 @@ func init() { flags.FVarP(cmdFlags, &Opt.ConflictResolve, "conflict-resolve", "", "Automatically resolve conflicts by preferring the version that is: "+ConflictResolveList+" (default: none)", "") flags.FVarP(cmdFlags, &Opt.ConflictLoser, "conflict-loser", "", "Action to take on the loser of a sync conflict (when there is a winner) or on both files (when there is no winner): "+ConflictLoserList+" (default: num)", "") flags.StringVarP(cmdFlags, &Opt.ConflictSuffixFlag, "conflict-suffix", "", Opt.ConflictSuffixFlag, "Suffix to use when renaming a --conflict-loser. Can be either one string or two comma-separated strings to assign different suffixes to Path1/Path2. (default: 'conflict')", "") + flags.BoolVarP(cmdFlags, &Opt.DeltaList, "delta-list", "", Opt.DeltaList, "Use Dropbox list_folder cursors for Path listings instead of a full walk. Local side still walks. Cursor+listing stored in --workdir.", "") _ = cmdFlags.MarkHidden("debugname") _ = cmdFlags.MarkHidden("localtime") addRC() diff --git a/cmd/bisync/delta_list.go b/cmd/bisync/delta_list.go new file mode 100644 index 000000000..b559f0e5b --- /dev/null +++ b/cmd/bisync/delta_list.go @@ -0,0 +1,152 @@ +package bisync + +import ( + "bufio" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// errDeltaTooManyDeletes is returned when applying a delta would exceed --max-delete. +var errDeltaTooManyDeletes = errors.New("delta list: too many deletes") + +// errDeltaCursorReset is returned when the remote invalidated the listing cursor. +// The in-memory listing must be left unchanged so the caller can full-relist. +var errDeltaCursorReset = errors.New("delta list: cursor reset") + +type listingDelta struct { + Remote string + Deleted bool + Size int64 + ModTime time.Time + Hash string + ID string + Flags string +} + +type deltaApplyOpts struct { + MaxDelete int + Force bool + Reset bool +} + +func cursorPath(listing string) string { + return listing + ".cursor" +} + +// applyListingDeltas mutates ls in place. On error, ls is restored to its +// previous contents (strong safety: never persist a half-applied delta). +func applyListingDeltas(ls *fileList, deltas []listingDelta, opt deltaApplyOpts) error { + if opt.Reset { + return errDeltaCursorReset + } + backupList := append([]string(nil), ls.list...) + backupInfo := make(map[string]*fileInfo, len(ls.info)) + for k, v := range ls.info { + cp := *v + backupInfo[k] = &cp + } + restore := func() { + ls.list = backupList + ls.info = backupInfo + } + + oldCount := 0 + for _, name := range ls.list { + if !ls.isDir(name) { + oldCount++ + } + } + deletedFiles := 0 + + for _, d := range deltas { + if d.Deleted { + // Dropbox DeletedMetadata: remove the path and all children. + prefix := d.Remote + if prefix != "" && !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + var toRemove []string + for _, name := range ls.list { + if name == d.Remote || (prefix != "" && strings.HasPrefix(name, prefix)) { + toRemove = append(toRemove, name) + } + } + for _, name := range toRemove { + if !ls.isDir(name) { + deletedFiles++ + } + ls.remove(name) + } + continue + } + flags := d.Flags + if flags == "" { + flags = "-" + } + ls.put(d.Remote, d.Size, d.ModTime, d.Hash, d.ID, flags) + } + + if !opt.Force && opt.MaxDelete >= 0 && oldCount > 0 { + maxRatio := float64(opt.MaxDelete) / 100.0 + if float64(deletedFiles)/float64(oldCount) > maxRatio { + restore() + return errDeltaTooManyDeletes + } + } + return nil +} + +func saveListingAndCursor(listing, cursor, root string, ls *fileList) error { + if err := os.MkdirAll(filepath.Dir(listing), 0700); err != nil { + return err + } + tmpListing := listing + ".tmp" + tmpCursor := cursorPath(listing) + ".tmp" + if err := ls.save(tmpListing); err != nil { + return err + } + cur := fmt.Sprintf("# rclone-bisync-cursor v1\nroot=%s\ncursor=%s\n", root, cursor) + if err := os.WriteFile(tmpCursor, []byte(cur), 0600); err != nil { + _ = os.Remove(tmpListing) + return err + } + if err := os.Rename(tmpListing, listing); err != nil { + _ = os.Remove(tmpListing) + _ = os.Remove(tmpCursor) + return err + } + if err := os.Rename(tmpCursor, cursorPath(listing)); err != nil { + return fmt.Errorf("listing saved but cursor rename failed (next run will full-list): %w", err) + } + return nil +} + +func loadCursor(listing string) (cursor, root string, err error) { + b, err := os.ReadFile(cursorPath(listing)) + if err != nil { + if os.IsNotExist(err) { + return "", "", nil + } + return "", "", err + } + sc := bufio.NewScanner(strings.NewReader(string(b))) + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "root=") { + root = strings.TrimPrefix(line, "root=") + } + if strings.HasPrefix(line, "cursor=") { + cursor = strings.TrimPrefix(line, "cursor=") + } + } + return cursor, root, sc.Err() +} + +func loadListingFile(listing string) (*fileList, error) { + b := &bisyncRun{} + return b.loadListing(listing) +} diff --git a/cmd/bisync/delta_list_run.go b/cmd/bisync/delta_list_run.go new file mode 100644 index 000000000..80e5f0411 --- /dev/null +++ b/cmd/bisync/delta_list_run.go @@ -0,0 +1,182 @@ +package bisync + +import ( + "context" + "errors" + "fmt" + + "github.com/rclone/rclone/backend/dropbox" + "github.com/rclone/rclone/fs" + "github.com/rclone/rclone/fs/hash" + "github.com/rclone/rclone/fs/walk" + "github.com/rclone/rclone/lib/terminal" +) + +func unwrapDropbox(f fs.Fs) *dropbox.Fs { + for f != nil { + if d, ok := f.(*dropbox.Fs); ok { + return d + } + uw := f.Features().UnWrap + if uw == nil { + return nil + } + f = uw() + } + return nil +} + +func (b *bisyncRun) makeDeltaListing(ctx context.Context) (*fileList, *fileList, error) { + b.march.marchCtx = ctx + b.setupListing() + + dbx1 := unwrapDropbox(b.fs1) + dbx2 := unwrapDropbox(b.fs2) + if dbx1 == nil && dbx2 == nil { + return nil, nil, errors.New("--delta-list requires a Dropbox remote on Path1 and/or Path2") + } + + var err error + if dbx1 != nil { + fs.Infof(nil, "Path1 Dropbox delta listing") + b.march.ls1, err = b.refreshDropboxListing(ctx, dbx1, b.fs1, b.listing1, b.march.ls1) + } else { + fs.Infof(nil, "Path1 full listing (no USN/journal; local walk)") + err = b.listOneFs(ctx, true) + } + if err != nil { + b.handleErr("delta-list", "error listing Path1", err, true, true) + b.abort = true + return b.march.ls1, b.march.ls2, err + } + + if dbx2 != nil { + fs.Infof(nil, "Path2 Dropbox delta listing") + b.march.ls2, err = b.refreshDropboxListing(ctx, dbx2, b.fs2, b.listing2, b.march.ls2) + } else { + fs.Infof(nil, "Path2 full listing (no USN/journal; local walk)") + err = b.listOneFs(ctx, false) + } + if err != nil { + b.handleErr("delta-list", "error listing Path2", err, true, true) + b.abort = true + return b.march.ls1, b.march.ls2, err + } + + if err = b.march.ls1.save(b.newListing1); err != nil { + return b.march.ls1, b.march.ls2, err + } + if err = b.march.ls2.save(b.newListing2); err != nil { + return b.march.ls1, b.march.ls2, err + } + return b.march.ls1, b.march.ls2, nil +} + +func (b *bisyncRun) listOneFs(ctx context.Context, isPath1 bool) error { + f := b.fs1 + if !isPath1 { + f = b.fs2 + } + return walk.ListR(ctx, f, "", false, -1, walk.ListAll, func(entries fs.DirEntries) error { + for _, e := range entries { + b.parse(e, isPath1) + } + return nil + }) +} + +func (b *bisyncRun) refreshDropboxListing(ctx context.Context, dbx *dropbox.Fs, f fs.Fs, listing string, ls *fileList) (*fileList, error) { + ls.hash = b.hashTypeForFs(f) + cursor, root, err := loadCursor(listing) + if err != nil { + return nil, err + } + wantRoot := bilibFsRoot(f) + if cursor != "" && root != "" && root != wantRoot { + fs.Infof(nil, "Dropbox cursor root %q != %q; full relist", root, wantRoot) + cursor = "" + } + + if cursor != "" { + prior, err := b.loadListing(listing) + if err != nil { + fs.Infof(nil, "No usable prior listing (%v); full Dropbox relist", err) + cursor = "" + } else { + ls = prior + ls.hash = b.hashTypeForFs(f) + } + } + + newCursor, reset, changes, err := dbx.ListFolderDeltas(ctx, cursor) + if err != nil { + return nil, err + } + if reset { + fs.Log(nil, Color(terminal.YellowFg, "Dropbox listing cursor reset; performing full recursive relist")) + newCursor, reset, changes, err = dbx.ListFolderDeltas(ctx, "") + if err != nil { + return nil, err + } + if reset { + return nil, errDeltaCursorReset + } + ls = newFileList() + ls.hash = b.hashTypeForFs(f) + } else if cursor == "" { + ls = newFileList() + ls.hash = b.hashTypeForFs(f) + } + + deltas := make([]listingDelta, 0, len(changes)) + for _, ch := range changes { + flags := "-" + if ch.IsDir { + flags = "d" + } + size := ch.Size + if ch.IsDir { + size = -1 + } + hashVal := ch.Hash + if ls.hash == hash.None { + hashVal = "" + } + deltas = append(deltas, listingDelta{ + Remote: ch.Remote, + Deleted: ch.Deleted, + Size: size, + ModTime: ch.ModTime.In(TZ), + Hash: hashVal, + Flags: flags, + }) + } + + applyErr := applyListingDeltas(ls, deltas, deltaApplyOpts{ + MaxDelete: b.opt.MaxDelete, + Force: b.opt.Force, + }) + if applyErr != nil { + return nil, applyErr + } + + if b.opt.DryRun { + fs.Infof(nil, "dry-run: not saving Dropbox listing cursor") + return ls, nil + } + if err := saveListingAndCursor(listing, newCursor, wantRoot, ls); err != nil { + return nil, err + } + return ls, nil +} + +func (b *bisyncRun) hashTypeForFs(f fs.Fs) hash.Type { + if f == b.fs1 { + return b.opt.Compare.HashType1 + } + return b.opt.Compare.HashType2 +} + +func bilibFsRoot(f fs.Fs) string { + return fmt.Sprintf("%s:%s", f.Name(), f.Root()) +} diff --git a/cmd/bisync/delta_list_test.go b/cmd/bisync/delta_list_test.go new file mode 100644 index 000000000..c86bf85b4 --- /dev/null +++ b/cmd/bisync/delta_list_test.go @@ -0,0 +1,122 @@ +package bisync + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/rclone/rclone/fs/hash" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func testListing() *fileList { + ls := newFileList() + ls.hash = hash.MD5 + t0 := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + ls.put("keep.txt", 10, t0, "aaa", "", "-") + ls.put("gone.txt", 20, t0, "bbb", "", "-") + ls.put("dir/a.txt", 30, t0, "ccc", "", "-") + ls.put("dir/b.txt", 40, t0, "ddd", "", "-") + ls.put("dir", 0, t0, "", "", "d") + return ls +} + +func TestApplyListingDeltas_UpsertAndDelete(t *testing.T) { + ls := testListing() + t1 := time.Date(2024, 2, 1, 0, 0, 0, 0, time.UTC) + err := applyListingDeltas(ls, []listingDelta{ + {Remote: "keep.txt", Size: 11, ModTime: t1, Hash: "zzz", Flags: "-"}, + {Remote: "new.txt", Size: 5, ModTime: t1, Hash: "eee", Flags: "-"}, + {Remote: "gone.txt", Deleted: true}, + }, deltaApplyOpts{MaxDelete: 50}) + require.NoError(t, err) + assert.True(t, ls.has("keep.txt")) + assert.Equal(t, int64(11), ls.getSize("keep.txt")) + assert.Equal(t, "zzz", ls.getHash("keep.txt")) + assert.True(t, ls.has("new.txt")) + assert.False(t, ls.has("gone.txt")) + assert.True(t, ls.has("dir/a.txt")) +} + +func TestApplyListingDeltas_FolderDeleteRemovesChildren(t *testing.T) { + ls := testListing() + err := applyListingDeltas(ls, []listingDelta{ + {Remote: "dir", Deleted: true, Flags: "d"}, + }, deltaApplyOpts{MaxDelete: 90}) + require.NoError(t, err) + assert.False(t, ls.has("dir")) + assert.False(t, ls.has("dir/a.txt")) + assert.False(t, ls.has("dir/b.txt")) + assert.True(t, ls.has("keep.txt")) +} + +func TestApplyListingDeltas_MaxDeleteAbortsWithoutMutation(t *testing.T) { + ls := testListing() + orig := len(ls.list) + err := applyListingDeltas(ls, []listingDelta{ + {Remote: "keep.txt", Deleted: true}, + {Remote: "gone.txt", Deleted: true}, + {Remote: "dir/a.txt", Deleted: true}, + }, deltaApplyOpts{MaxDelete: 50}) + require.Error(t, err) + assert.True(t, errors.Is(err, errDeltaTooManyDeletes)) + assert.Equal(t, orig, len(ls.list)) + assert.True(t, ls.has("keep.txt")) + assert.True(t, ls.has("gone.txt")) +} + +func TestApplyListingDeltas_ForceBypassesMaxDelete(t *testing.T) { + ls := testListing() + err := applyListingDeltas(ls, []listingDelta{ + {Remote: "keep.txt", Deleted: true}, + {Remote: "gone.txt", Deleted: true}, + {Remote: "dir/a.txt", Deleted: true}, + }, deltaApplyOpts{MaxDelete: 50, Force: true}) + require.NoError(t, err) + assert.False(t, ls.has("keep.txt")) +} + +func TestApplyListingDeltas_ResetLeavesListingUntouched(t *testing.T) { + ls := testListing() + orig := len(ls.list) + err := applyListingDeltas(ls, []listingDelta{ + {Remote: "keep.txt", Deleted: true}, + }, deltaApplyOpts{Reset: true, MaxDelete: 50}) + require.Error(t, err) + assert.True(t, errors.Is(err, errDeltaCursorReset)) + assert.Equal(t, orig, len(ls.list)) + assert.True(t, ls.has("keep.txt")) +} + +func TestAtomicCursorAndListing(t *testing.T) { + dir := t.TempDir() + listing := filepath.Join(dir, "path2.lst") + ls := testListing() + require.NoError(t, saveListingAndCursor(listing, "CUR1", "/Dropbox", ls)) + got, root, err := loadCursor(listing) + require.NoError(t, err) + assert.Equal(t, "CUR1", got) + assert.Equal(t, "/Dropbox", root) + ls2, err := loadListingFile(listing) + require.NoError(t, err) + assert.True(t, ls2.has("keep.txt")) + + require.NoError(t, saveListingAndCursor(listing, "CUR2", "/Dropbox", ls)) + got, _, err = loadCursor(listing) + require.NoError(t, err) + assert.Equal(t, "CUR2", got) +} + +func TestLoadCursor_MissingIsEmpty(t *testing.T) { + dir := t.TempDir() + listing := filepath.Join(dir, "nope.lst") + got, root, err := loadCursor(listing) + require.NoError(t, err) + assert.Empty(t, got) + assert.Empty(t, root) + _, err = os.Stat(cursorPath(listing)) + assert.True(t, os.IsNotExist(err)) +} diff --git a/cmd/bisync/march.go b/cmd/bisync/march.go index 10e22dc38..bbe33edc9 100644 --- a/cmd/bisync/march.go +++ b/cmd/bisync/march.go @@ -24,6 +24,9 @@ type bisyncMarch struct { } func (b *bisyncRun) makeMarchListing(ctx context.Context) (*fileList, *fileList, error) { + if b.opt.DeltaList { + return b.makeDeltaListing(ctx) + } ci := fs.GetConfig(ctx) b.march.marchCtx = ctx b.setupListing() diff --git a/cmd/bisync/rc.go b/cmd/bisync/rc.go index 82be54185..c49d0eabb 100644 --- a/cmd/bisync/rc.go +++ b/cmd/bisync/rc.go @@ -146,6 +146,9 @@ func rcBisync(ctx context.Context, in rc.Params) (out rc.Params, err error) { if opt.Recover, err = in.GetBool("recover"); rc.NotErrParamNotFound(err) { fs.Debugf("recover", "optional parameter is missing. using default value: %v", opt.Recover) } + if opt.DeltaList, err = in.GetBool("deltaList"); rc.NotErrParamNotFound(err) { + fs.Debugf("deltaList", "optional parameter is missing. using default value: %v", opt.DeltaList) + } if opt.CompareFlag, err = in.GetString("compare"); rc.NotErrParamNotFound(err) { fs.Debugf("compare", "optional parameter is missing. using default value: %v", opt.CompareFlag) }