Merge pull request #1540 from calmh/conflicts

Handle conflicts when pulling (fixes #220)

internal/model/rwfolder.go

@@ -588,7 +588,11 @@ func (p *rwFolder) deleteFile(file protocol.FileInfo) {
 
 	realName := filepath.Join(p.dir, file.Name)
 
-	if p.versioner != nil {
+	cur, ok := p.model.CurrentFolderFile(p.folder, file.Name)
+	if ok && cur.Version.Concurrent(file.Version) {
+		// There is a conflict here. Move the file to a conflict copy instead of deleting.
+		err = osutil.InWritableDir(moveForConflict, realName)
+	} else if p.versioner != nil {
 		err = osutil.InWritableDir(p.versioner.Archive, realName)
 	} else {
 		err = osutil.InWritableDir(os.Remove, realName)
@@ -756,6 +760,7 @@ func (p *rwFolder) handleFile(file protocol.FileInfo, copyChan chan<- copyBlocks
 		copyNeeded:  len(blocks),
 		reused:      reused,
 		ignorePerms: p.ignorePerms,
+		version:     curFile.Version,
 	}
 
 	if debug {
@@ -966,6 +971,7 @@ func (p *rwFolder) performFinish(state *sharedPullerState) {
 			"error":  err,
 		})
 	}()
+
 	// Set the correct permission bits on the new file
 	if !p.ignorePerms {
 		err = os.Chmod(state.tempName, os.FileMode(state.file.Flags&0777))
@@ -991,15 +997,22 @@ func (p *rwFolder) performFinish(state *sharedPullerState) {
 		}
 	}
 
-	// If we should use versioning, let the versioner archive the old
-	// file before we replace it. Archiving a non-existent file is not
-	// an error.
-	if p.versioner != nil {
+	if state.version.Concurrent(state.file.Version) {
+		// The new file has been changed in conflict with the existing one. We
+		// should file it away as a conflict instead of just removing or
+		// archiving.
+		err = osutil.InWritableDir(moveForConflict, state.realName)
+	} else if p.versioner != nil {
+		// If we should use versioning, let the versioner archive the old
+		// file before we replace it. Archiving a non-existent file is not
+		// an error.
 		err = p.versioner.Archive(state.realName)
-		if err != nil {
-			l.Warnln("Puller: final:", err)
-			return
-		}
+	} else {
+		err = nil
+	}
+	if err != nil {
+		l.Warnln("Puller: final:", err)
+		return
 	}
 
 	// If the target path is a symlink or a directory, we cannot copy
@@ -1095,3 +1108,10 @@ func removeDevice(devices []protocol.DeviceID, device protocol.DeviceID) []proto
 	}
 	return devices
 }
+
+func moveForConflict(name string) error {
+	ext := filepath.Ext(name)
+	withoutExt := name[:len(name)-len(ext)]
+	newName := withoutExt + time.Now().Format(".sync-conflict-20060102-150405") + ext
+	return os.Rename(name, newName)
+}

internal/model/sharedpullerstate.go

@@ -20,12 +20,13 @@ import (
 // updated along the way.
 type sharedPullerState struct {
 	// Immutable, does not require locking
-	file        protocol.FileInfo
+	file        protocol.FileInfo // The new file (desired end state)
 	folder      string
 	tempName    string
 	realName    string
 	reused      int // Number of blocks reused from temporary file
 	ignorePerms bool
+	version     protocol.Vector // The current (old) version
 
 	// Mutable, must be locked for access
 	err        error      // The first error we hit

test/conflict_test.go

@@ -0,0 +1,247 @@
+// Copyright (C) 2015 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build integration
+
+package integration
+
+import (
+	"log"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestConflict(t *testing.T) {
+	log.Println("Cleaning...")
+	err := removeAll("s1", "s2", "h1/index", "h2/index")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	log.Println("Generating files...")
+	err = generateFiles("s1", 100, 20, "../LICENSE")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fd, err := os.Create("s1/testfile.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = fd.WriteString("hello\n")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = fd.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expected, err := directoryContents("s1")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	log.Println("Starting sender...")
+	sender := syncthingProcess{ // id1
+		instance: "1",
+		argv:     []string{"-home", "h1"},
+		port:     8081,
+		apiKey:   apiKey,
+	}
+	err = sender.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sender.stop()
+
+	// Wait for one scan to succeed, or up to 20 seconds... This is to let
+	// startup, UPnP etc complete and make sure the sender has the full index
+	// before they connect.
+	for i := 0; i < 20; i++ {
+		resp, err := sender.post("/rest/scan?folder=default", nil)
+		if err != nil {
+			time.Sleep(time.Second)
+			continue
+		}
+		if resp.StatusCode != 200 {
+			resp.Body.Close()
+			time.Sleep(time.Second)
+			continue
+		}
+		break
+	}
+
+	log.Println("Starting receiver...")
+	receiver := syncthingProcess{ // id2
+		instance: "2",
+		argv:     []string{"-home", "h2"},
+		port:     8082,
+		apiKey:   apiKey,
+	}
+	err = receiver.start()
+	if err != nil {
+		sender.stop()
+		t.Fatal(err)
+	}
+	defer receiver.stop()
+
+	if err = coCompletion(sender, receiver); err != nil {
+		t.Fatal(err)
+	}
+
+	sender.stop()
+	receiver.stop()
+
+	log.Println("Verifying...")
+
+	actual, err := directoryContents("s2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = compareDirectoryContents(actual, expected)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	log.Println("Introducing a conflict (simultaneous edit)...")
+
+	fd, err = os.OpenFile("s1/testfile.txt", os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = fd.WriteString("text added to s1\n")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = fd.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fd, err = os.OpenFile("s2/testfile.txt", os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = fd.WriteString("text added to s2\n")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = fd.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	log.Println("Syncing...")
+
+	err = receiver.start()
+	err = sender.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err != nil {
+		sender.stop()
+		t.Fatal(err)
+	}
+
+	if err = coCompletion(sender, receiver); err != nil {
+		t.Fatal(err)
+	}
+
+	sender.stop()
+	receiver.stop()
+
+	// The conflict is expected on the s2 side due to how we calculate which
+	// file is the winner (based on device ID)
+
+	files, err := filepath.Glob("s2/*sync-conflict*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(files) != 1 {
+		t.Errorf("Expected 1 conflicted files instead of %d", len(files))
+	}
+
+	log.Println("Introducing a conflict (edit plus delete)...")
+
+	err = os.Remove("s1/testfile.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fd, err = os.OpenFile("s2/testfile.txt", os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = fd.WriteString("more text added to s2\n")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = fd.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	log.Println("Syncing...")
+
+	err = receiver.start()
+	err = sender.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err != nil {
+		sender.stop()
+		t.Fatal(err)
+	}
+
+	if err = coCompletion(sender, receiver); err != nil {
+		t.Fatal(err)
+	}
+
+	sender.stop()
+	receiver.stop()
+
+	// The conflict should manifest on the s2 side again, where we should have
+	// moved the file to a conflict copy instead of just deleting it.
+
+	files, err = filepath.Glob("s2/*sync-conflict*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(files) != 2 {
+		t.Errorf("Expected 2 conflicted files instead of %d", len(files))
+	}
+}
+
+func coCompletion(p ...syncthingProcess) error {
+mainLoop:
+	for {
+		time.Sleep(2500 * time.Millisecond)
+
+		tot := 0
+		for i := range p {
+			comp, err := p[i].peerCompletion()
+			if err != nil {
+				if isTimeout(err) {
+					continue mainLoop
+				}
+				return err
+			}
+
+			for _, pct := range comp {
+				tot += pct
+			}
+		}
+
+		if tot == 100*(len(p)) {
+			return nil
+		}
+
+		log.Printf("%d / %d...", tot, 100*(len(p)))
+	}
+}