mirror of
https://github.com/rclone/rclone.git
synced 2025-12-15 15:53:41 +00:00
Implement --compare-dest & --copy-dest Fixes #3278
This commit is contained in:
yparitcher
committed by
Nick Craig-Wood
Nick Craig-Wood
parent
266600dba7
commit
8e8b78d7e5
101
fs/sync/sync.go
101
fs/sync/sync.go
@@ -28,39 +28,40 @@ type syncCopyMove struct {
|
||||
deleteEmptySrcDirs bool
|
||||
dir string
|
||||
// internal state
|
||||
ctx context.Context // internal context for controlling go-routines
|
||||
cancel func() // cancel the context
|
||||
noTraverse bool // if set don't traverse the dst
|
||||
deletersWg sync.WaitGroup // for delete before go routine
|
||||
deleteFilesCh chan fs.Object // channel to receive deletes if delete before
|
||||
trackRenames bool // set if we should do server side renames
|
||||
dstFilesMu sync.Mutex // protect dstFiles
|
||||
dstFiles map[string]fs.Object // dst files, always filled
|
||||
srcFiles map[string]fs.Object // src files, only used if deleteBefore
|
||||
srcFilesChan chan fs.Object // passes src objects
|
||||
srcFilesResult chan error // error result of src listing
|
||||
dstFilesResult chan error // error result of dst listing
|
||||
dstEmptyDirsMu sync.Mutex // protect dstEmptyDirs
|
||||
dstEmptyDirs map[string]fs.DirEntry // potentially empty directories
|
||||
srcEmptyDirsMu sync.Mutex // protect srcEmptyDirs
|
||||
srcEmptyDirs map[string]fs.DirEntry // potentially empty directories
|
||||
checkerWg sync.WaitGroup // wait for checkers
|
||||
toBeChecked *pipe // checkers channel
|
||||
transfersWg sync.WaitGroup // wait for transfers
|
||||
toBeUploaded *pipe // copiers channel
|
||||
errorMu sync.Mutex // Mutex covering the errors variables
|
||||
err error // normal error from copy process
|
||||
noRetryErr error // error with NoRetry set
|
||||
fatalErr error // fatal error
|
||||
commonHash hash.Type // common hash type between src and dst
|
||||
renameMapMu sync.Mutex // mutex to protect the below
|
||||
renameMap map[string][]fs.Object // dst files by hash - only used by trackRenames
|
||||
renamerWg sync.WaitGroup // wait for renamers
|
||||
toBeRenamed *pipe // renamers channel
|
||||
trackRenamesWg sync.WaitGroup // wg for background track renames
|
||||
trackRenamesCh chan fs.Object // objects are pumped in here
|
||||
renameCheck []fs.Object // accumulate files to check for rename here
|
||||
backupDir fs.Fs // place to store overwrites/deletes
|
||||
ctx context.Context // internal context for controlling go-routines
|
||||
cancel func() // cancel the context
|
||||
noTraverse bool // if set don't traverse the dst
|
||||
deletersWg sync.WaitGroup // for delete before go routine
|
||||
deleteFilesCh chan fs.Object // channel to receive deletes if delete before
|
||||
trackRenames bool // set if we should do server side renames
|
||||
dstFilesMu sync.Mutex // protect dstFiles
|
||||
dstFiles map[string]fs.Object // dst files, always filled
|
||||
srcFiles map[string]fs.Object // src files, only used if deleteBefore
|
||||
srcFilesChan chan fs.Object // passes src objects
|
||||
srcFilesResult chan error // error result of src listing
|
||||
dstFilesResult chan error // error result of dst listing
|
||||
dstEmptyDirsMu sync.Mutex // protect dstEmptyDirs
|
||||
dstEmptyDirs map[string]fs.DirEntry // potentially empty directories
|
||||
srcEmptyDirsMu sync.Mutex // protect srcEmptyDirs
|
||||
srcEmptyDirs map[string]fs.DirEntry // potentially empty directories
|
||||
checkerWg sync.WaitGroup // wait for checkers
|
||||
toBeChecked *pipe // checkers channel
|
||||
transfersWg sync.WaitGroup // wait for transfers
|
||||
toBeUploaded *pipe // copiers channel
|
||||
errorMu sync.Mutex // Mutex covering the errors variables
|
||||
err error // normal error from copy process
|
||||
noRetryErr error // error with NoRetry set
|
||||
fatalErr error // fatal error
|
||||
commonHash hash.Type // common hash type between src and dst
|
||||
renameMapMu sync.Mutex // mutex to protect the below
|
||||
renameMap map[string][]fs.Object // dst files by hash - only used by trackRenames
|
||||
renamerWg sync.WaitGroup // wait for renamers
|
||||
toBeRenamed *pipe // renamers channel
|
||||
trackRenamesWg sync.WaitGroup // wg for background track renames
|
||||
trackRenamesCh chan fs.Object // objects are pumped in here
|
||||
renameCheck []fs.Object // accumulate files to check for rename here
|
||||
compareCopyDest fs.Fs // place to check for files to server side copy
|
||||
backupDir fs.Fs // place to store overwrites/deletes
|
||||
}
|
||||
|
||||
func newSyncCopyMove(ctx context.Context, fdst, fsrc fs.Fs, deleteMode fs.DeleteMode, DoMove bool, deleteEmptySrcDirs bool, copyEmptySrcDirs bool) (*syncCopyMove, error) {
|
||||
@@ -127,6 +128,19 @@ func newSyncCopyMove(ctx context.Context, fdst, fsrc fs.Fs, deleteMode fs.Delete
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if fs.Config.CompareDest != "" {
|
||||
var err error
|
||||
s.compareCopyDest, err = operations.GetCompareDest()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else if fs.Config.CopyDest != "" {
|
||||
var err error
|
||||
s.compareCopyDest, err = operations.GetCopyDest(fdst)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
@@ -204,7 +218,11 @@ func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, wg *sync.WaitGroup) {
|
||||
accounting.Stats.Checking(src.Remote())
|
||||
// Check to see if can store this
|
||||
if src.Storable() {
|
||||
if operations.NeedTransfer(s.ctx, pair.Dst, pair.Src) {
|
||||
NoNeedTransfer, err := operations.CompareOrCopyDest(s.ctx, s.fdst, pair.Dst, pair.Src, s.compareCopyDest, s.backupDir)
|
||||
if err != nil {
|
||||
s.processError(err)
|
||||
}
|
||||
if !NoNeedTransfer && operations.NeedTransfer(s.ctx, pair.Dst, pair.Src) {
|
||||
// If files are treated as immutable, fail if destination exists and does not match
|
||||
if fs.Config.Immutable && pair.Dst != nil {
|
||||
fs.Errorf(pair.Dst, "Source and destination exist but do not match: immutable file modified")
|
||||
@@ -764,10 +782,17 @@ func (s *syncCopyMove) SrcOnly(src fs.DirEntry) (recurse bool) {
|
||||
case s.trackRenamesCh <- x:
|
||||
}
|
||||
} else {
|
||||
// No need to check since doesn't exist
|
||||
ok := s.toBeUploaded.Put(s.ctx, fs.ObjectPair{Src: x, Dst: nil})
|
||||
if !ok {
|
||||
return
|
||||
// Check CompareDest && CopyDest
|
||||
NoNeedTransfer, err := operations.CompareOrCopyDest(s.ctx, s.fdst, nil, x, s.compareCopyDest, s.backupDir)
|
||||
if err != nil {
|
||||
s.processError(err)
|
||||
}
|
||||
if !NoNeedTransfer {
|
||||
// No need to check since doesn't exist
|
||||
ok := s.toBeUploaded.Put(s.ctx, fs.ObjectPair{Src: x, Dst: nil})
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
case fs.Directory:
|
||||
|
||||
@@ -1216,6 +1216,196 @@ func TestSyncOverlap(t *testing.T) {
|
||||
checkErr(Sync(context.Background(), FremoteSync, FremoteSync, false))
|
||||
}
|
||||
|
||||
// Test with CompareDest set
|
||||
func TestSyncCompareDest(t *testing.T) {
|
||||
r := fstest.NewRun(t)
|
||||
defer r.Finalise()
|
||||
|
||||
fs.Config.CompareDest = r.FremoteName + "/CompareDest"
|
||||
defer func() {
|
||||
fs.Config.CompareDest = ""
|
||||
}()
|
||||
|
||||
fdst, err := fs.NewFs(r.FremoteName + "/dst")
|
||||
require.NoError(t, err)
|
||||
|
||||
// check empty dest, empty compare
|
||||
file1 := r.WriteFile("one", "one", t1)
|
||||
fstest.CheckItems(t, r.Flocal, file1)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file1dst := file1
|
||||
file1dst.Path = "dst/one"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file1dst)
|
||||
|
||||
// check old dest, empty compare
|
||||
file1b := r.WriteFile("one", "onet2", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file1dst)
|
||||
fstest.CheckItems(t, r.Flocal, file1b)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file1bdst := file1b
|
||||
file1bdst.Path = "dst/one"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file1bdst)
|
||||
|
||||
// check old dest, new compare
|
||||
file3 := r.WriteObject(context.Background(), "dst/one", "one", t1)
|
||||
file2 := r.WriteObject(context.Background(), "CompareDest/one", "onet2", t2)
|
||||
file1c := r.WriteFile("one", "onet2", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3)
|
||||
fstest.CheckItems(t, r.Flocal, file1c)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3)
|
||||
|
||||
// check empty dest, new compare
|
||||
file4 := r.WriteObject(context.Background(), "CompareDest/two", "two", t2)
|
||||
file5 := r.WriteFile("two", "two", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3, file4)
|
||||
fstest.CheckItems(t, r.Flocal, file1c, file5)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3, file4)
|
||||
|
||||
// check new dest, new compare
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3, file4)
|
||||
|
||||
// check empty dest, old compare
|
||||
file5b := r.WriteFile("two", "twot3", t3)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3, file4)
|
||||
fstest.CheckItems(t, r.Flocal, file1c, file5b)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file5bdst := file5b
|
||||
file5bdst.Path = "dst/two"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3, file4, file5bdst)
|
||||
}
|
||||
|
||||
// Test with CopyDest set
|
||||
func TestSyncCopyDest(t *testing.T) {
|
||||
r := fstest.NewRun(t)
|
||||
defer r.Finalise()
|
||||
|
||||
if r.Fremote.Features().Copy == nil {
|
||||
t.Skip("Skipping test as remote does not support server side copy")
|
||||
}
|
||||
|
||||
fs.Config.CopyDest = r.FremoteName + "/CopyDest"
|
||||
defer func() {
|
||||
fs.Config.CopyDest = ""
|
||||
}()
|
||||
|
||||
fdst, err := fs.NewFs(r.FremoteName + "/dst")
|
||||
require.NoError(t, err)
|
||||
|
||||
// check empty dest, empty copy
|
||||
file1 := r.WriteFile("one", "one", t1)
|
||||
fstest.CheckItems(t, r.Flocal, file1)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file1dst := file1
|
||||
file1dst.Path = "dst/one"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file1dst)
|
||||
|
||||
// check old dest, empty copy
|
||||
file1b := r.WriteFile("one", "onet2", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file1dst)
|
||||
fstest.CheckItems(t, r.Flocal, file1b)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file1bdst := file1b
|
||||
file1bdst.Path = "dst/one"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file1bdst)
|
||||
|
||||
// check old dest, new copy, backup-dir
|
||||
|
||||
fs.Config.BackupDir = r.FremoteName + "/BackupDir"
|
||||
|
||||
file3 := r.WriteObject(context.Background(), "dst/one", "one", t1)
|
||||
file2 := r.WriteObject(context.Background(), "CopyDest/one", "onet2", t2)
|
||||
file1c := r.WriteFile("one", "onet2", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file3)
|
||||
fstest.CheckItems(t, r.Flocal, file1c)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file2dst := file2
|
||||
file2dst.Path = "dst/one"
|
||||
file3.Path = "BackupDir/one"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3)
|
||||
fs.Config.BackupDir = ""
|
||||
|
||||
// check empty dest, new copy
|
||||
file4 := r.WriteObject(context.Background(), "CopyDest/two", "two", t2)
|
||||
file5 := r.WriteFile("two", "two", t2)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3, file4)
|
||||
fstest.CheckItems(t, r.Flocal, file1c, file5)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file4dst := file4
|
||||
file4dst.Path = "dst/two"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3, file4, file4dst)
|
||||
|
||||
// check new dest, new copy
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3, file4, file4dst)
|
||||
|
||||
// check empty dest, old copy
|
||||
file6 := r.WriteObject(context.Background(), "CopyDest/three", "three", t2)
|
||||
file7 := r.WriteFile("three", "threet3", t3)
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3, file4, file4dst, file6)
|
||||
fstest.CheckItems(t, r.Flocal, file1c, file5, file7)
|
||||
|
||||
accounting.Stats.ResetCounters()
|
||||
err = Sync(context.Background(), fdst, r.Flocal, false)
|
||||
require.NoError(t, err)
|
||||
|
||||
file7dst := file7
|
||||
file7dst.Path = "dst/three"
|
||||
|
||||
fstest.CheckItems(t, r.Fremote, file2, file2dst, file3, file4, file4dst, file6, file7dst)
|
||||
}
|
||||
|
||||
// Test with BackupDir set
|
||||
func testSyncBackupDir(t *testing.T, suffix string, suffixKeepExtension bool) {
|
||||
r := fstest.NewRun(t)
|
||||
|
||||
Reference in New Issue
Block a user