diff --git a/Godeps/Godeps.json b/Godeps/Godeps.json index 15857ef0..1886f33d 100644 --- a/Godeps/Godeps.json +++ b/Godeps/Godeps.json @@ -31,7 +31,7 @@ }, { "ImportPath": "github.com/syncthing/protocol", - "Rev": "3d8a71fdb205fe2401a341a739208bc9d1e79a1b" + "Rev": "e7db2648034fb71b051902a02bc25d4468ed492e" }, { "ImportPath": "github.com/syndtr/goleveldb/leveldb", diff --git a/Godeps/_workspace/src/github.com/syncthing/protocol/vector.go b/Godeps/_workspace/src/github.com/syncthing/protocol/vector.go index 04859452..edd15614 100644 --- a/Godeps/_workspace/src/github.com/syncthing/protocol/vector.go +++ b/Godeps/_workspace/src/github.com/syncthing/protocol/vector.go @@ -103,3 +103,13 @@ func (a Vector) Concurrent(b Vector) bool { comp := a.Compare(b) return comp == ConcurrentGreater || comp == ConcurrentLesser } + +// Counter returns the current value of the given counter ID. +func (v Vector) Counter(id uint64) uint64 { + for _, c := range v { + if c.ID == id { + return c.Value + } + } + return 0 +} diff --git a/Godeps/_workspace/src/github.com/syncthing/protocol/vector_test.go b/Godeps/_workspace/src/github.com/syncthing/protocol/vector_test.go index 7815412c..c01255e7 100644 --- a/Godeps/_workspace/src/github.com/syncthing/protocol/vector_test.go +++ b/Godeps/_workspace/src/github.com/syncthing/protocol/vector_test.go @@ -118,5 +118,17 @@ func TestMerge(t *testing.T) { t.Errorf("%d: %+v.Merge(%+v) == %+v (expected %+v)", i, tc.a, tc.b, m, tc.m) } } - +} + +func TestCounterValue(t *testing.T) { + v0 := Vector{Counter{42, 1}, Counter{64, 5}} + if v0.Counter(42) != 1 { + t.Error("Counter error, %d != %d", v0.Counter(42), 1) + } + if v0.Counter(64) != 5 { + t.Error("Counter error, %d != %d", v0.Counter(64), 5) + } + if v0.Counter(72) != 0 { + t.Error("Counter error, %d != %d", v0.Counter(72), 0) + } } diff --git a/Godeps/_workspace/src/github.com/thejerf/suture/pre-commit b/Godeps/_workspace/src/github.com/thejerf/suture/pre-commit index c88ec0fe..6199d610 100644 --- a/Godeps/_workspace/src/github.com/thejerf/suture/pre-commit +++ b/Godeps/_workspace/src/github.com/thejerf/suture/pre-commit @@ -9,3 +9,4 @@ if [ ! -z "$GOLINTOUT" -o "$?" != 0 ]; then fi go test + diff --git a/internal/model/model.go b/internal/model/model.go index fa122d8a..7965fca3 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -142,7 +142,7 @@ func (m *Model) StartFolderRW(folder string) { if ok { panic("cannot start already running folder " + folder) } - p := newRWFolder(m, cfg) + p := newRWFolder(m, m.shortID, cfg) m.folderRunners[folder] = p m.fmut.Unlock() diff --git a/internal/model/rwfolder.go b/internal/model/rwfolder.go index 97108165..25eafcf5 100644 --- a/internal/model/rwfolder.go +++ b/internal/model/rwfolder.go @@ -68,13 +68,14 @@ type rwFolder struct { lenientMtimes bool copiers int pullers int + shortID uint64 stop chan struct{} queue *jobQueue dbUpdates chan protocol.FileInfo } -func newRWFolder(m *Model, cfg config.FolderConfiguration) *rwFolder { +func newRWFolder(m *Model, shortID uint64, cfg config.FolderConfiguration) *rwFolder { return &rwFolder{ stateTracker: stateTracker{folder: cfg.ID}, @@ -88,6 +89,7 @@ func newRWFolder(m *Model, cfg config.FolderConfiguration) *rwFolder { lenientMtimes: cfg.LenientMtimes, copiers: cfg.Copiers, pullers: cfg.Pullers, + shortID: shortID, stop: make(chan struct{}), queue: newJobQueue(), @@ -603,8 +605,11 @@ func (p *rwFolder) deleteFile(file protocol.FileInfo) { realName := filepath.Join(p.dir, file.Name) 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. + if ok && p.inConflict(cur.Version, file.Version) { + // There is a conflict here. Move the file to a conflict copy instead + // of deleting. Also merge with the version vector we had, to indicate + // we have resolved the conflict. + file.Version = file.Version.Merge(cur.Version) err = osutil.InWritableDir(moveForConflict, realName) } else if p.versioner != nil { err = osutil.InWritableDir(p.versioner.Archive, realName) @@ -816,6 +821,12 @@ func (p *rwFolder) shortcutFile(file protocol.FileInfo) (err error) { } } + // This may have been a conflict. We should merge the version vectors so + // that our clock doesn't move backwards. + if cur, ok := p.model.CurrentFolderFile(p.folder, file.Name); ok { + file.Version = file.Version.Merge(cur.Version) + } + p.dbUpdates <- file return } @@ -1011,10 +1022,12 @@ func (p *rwFolder) performFinish(state *sharedPullerState) { } } - if state.version.Concurrent(state.file.Version) { + if p.inConflict(state.version, 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. + // archiving. Also merge with the version vector we had, to indicate + // we have resolved the conflict. + state.file.Version = state.file.Version.Merge(state.version) err = osutil.InWritableDir(moveForConflict, state.realName) } else if p.versioner != nil { // If we should use versioning, let the versioner archive the old @@ -1144,6 +1157,22 @@ loop: } } +func (p *rwFolder) inConflict(current, replacement protocol.Vector) bool { + if current.Concurrent(replacement) { + // Obvious case + return true + } + if replacement.Counter(p.shortID) > current.Counter(p.shortID) { + // The replacement file contains a higher version for ourselves than + // what we have. This isn't supposed to be possible, since it's only + // we who can increment that counter. We take it as a sign that + // something is wrong (our index may have been corrupted or removed) + // and flag it as a conflict. + return true + } + return false +} + func invalidateFolder(cfg *config.Configuration, folderID string, err error) { for i := range cfg.Folders { folder := &cfg.Folders[i]