fa0101bd60
This changes the BEP protocol to use protocol buffer serialization instead of XDR, and therefore also the database format. The local discovery protocol is also updated to be protocol buffer format. GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3276 LGTM: AudriusButkevicius
444 lines
11 KiB
Go
444 lines
11 KiB
Go
// Copyright (C) 2014 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/.
|
|
|
|
package model
|
|
|
|
import (
|
|
"fmt"
|
|
"path/filepath"
|
|
"runtime"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/syncthing/syncthing/lib/config"
|
|
"github.com/syncthing/syncthing/lib/events"
|
|
"github.com/syncthing/syncthing/lib/protocol"
|
|
"github.com/syncthing/syncthing/lib/sync"
|
|
)
|
|
|
|
var timeout = 100 * time.Millisecond
|
|
|
|
func caller(skip int) string {
|
|
_, file, line, ok := runtime.Caller(skip + 1)
|
|
if !ok {
|
|
return "unknown"
|
|
}
|
|
return fmt.Sprintf("%s:%d", filepath.Base(file), line)
|
|
}
|
|
|
|
func expectEvent(w *events.Subscription, t *testing.T, size int) {
|
|
event, err := w.Poll(timeout)
|
|
if err != nil {
|
|
t.Fatal("Unexpected error:", err, "at", caller(1))
|
|
}
|
|
if event.Type != events.DownloadProgress {
|
|
t.Fatal("Unexpected event:", event, "at", caller(1))
|
|
}
|
|
data := event.Data.(map[string]map[string]*pullerProgress)
|
|
if len(data) != size {
|
|
t.Fatal("Unexpected event data size:", data, "at", caller(1))
|
|
}
|
|
}
|
|
|
|
func expectTimeout(w *events.Subscription, t *testing.T) {
|
|
_, err := w.Poll(timeout)
|
|
if err != events.ErrTimeout {
|
|
t.Fatal("Unexpected non-Timeout error:", err, "at", caller(1))
|
|
}
|
|
}
|
|
|
|
func TestProgressEmitter(t *testing.T) {
|
|
w := events.Default.Subscribe(events.DownloadProgress)
|
|
|
|
c := config.Wrap("/tmp/test", config.Configuration{})
|
|
c.SetOptions(config.OptionsConfiguration{
|
|
ProgressUpdateIntervalS: 0,
|
|
})
|
|
|
|
p := NewProgressEmitter(c)
|
|
go p.Serve()
|
|
|
|
expectTimeout(w, t)
|
|
|
|
s := sharedPullerState{
|
|
updated: time.Now(),
|
|
mut: sync.NewRWMutex(),
|
|
}
|
|
p.Register(&s)
|
|
|
|
expectEvent(w, t, 1)
|
|
expectTimeout(w, t)
|
|
|
|
s.copyDone(protocol.BlockInfo{})
|
|
|
|
expectEvent(w, t, 1)
|
|
expectTimeout(w, t)
|
|
|
|
s.copiedFromOrigin()
|
|
|
|
expectEvent(w, t, 1)
|
|
expectTimeout(w, t)
|
|
|
|
s.pullStarted()
|
|
|
|
expectEvent(w, t, 1)
|
|
expectTimeout(w, t)
|
|
|
|
s.pullDone(protocol.BlockInfo{})
|
|
|
|
expectEvent(w, t, 1)
|
|
expectTimeout(w, t)
|
|
|
|
p.Deregister(&s)
|
|
|
|
expectEvent(w, t, 0)
|
|
expectTimeout(w, t)
|
|
|
|
}
|
|
|
|
func TestSendDownloadProgressMessages(t *testing.T) {
|
|
c := config.Wrap("/tmp/test", config.Configuration{})
|
|
c.SetOptions(config.OptionsConfiguration{
|
|
ProgressUpdateIntervalS: 0,
|
|
TempIndexMinBlocks: 10,
|
|
})
|
|
|
|
fc := &FakeConnection{}
|
|
|
|
p := NewProgressEmitter(c)
|
|
p.temporaryIndexSubscribe(fc, []string{"folder", "folder2"})
|
|
|
|
expect := func(updateIdx int, state *sharedPullerState, updateType protocol.FileDownloadProgressUpdateType, version protocol.Vector, blocks []int32, remove bool) {
|
|
messageIdx := -1
|
|
for i, msg := range fc.downloadProgressMessages {
|
|
if msg.folder == state.folder {
|
|
messageIdx = i
|
|
break
|
|
}
|
|
}
|
|
if messageIdx < 0 {
|
|
t.Errorf("Message for folder %s does not exist at %s", state.folder, caller(1))
|
|
}
|
|
|
|
msg := fc.downloadProgressMessages[messageIdx]
|
|
|
|
// Don't know the index (it's random due to iterating maps)
|
|
if updateIdx == -1 {
|
|
for i, upd := range msg.updates {
|
|
if upd.Name == state.file.Name {
|
|
updateIdx = i
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
if updateIdx == -1 {
|
|
t.Errorf("Could not find update for %s at %s", state.file.Name, caller(1))
|
|
}
|
|
|
|
if updateIdx > len(msg.updates)-1 {
|
|
t.Errorf("Update at index %d does not exist at %s", updateIdx, caller(1))
|
|
}
|
|
|
|
update := msg.updates[updateIdx]
|
|
|
|
if update.UpdateType != updateType {
|
|
t.Errorf("Wrong update type at %s", caller(1))
|
|
}
|
|
|
|
if !update.Version.Equal(version) {
|
|
t.Errorf("Wrong version at %s", caller(1))
|
|
}
|
|
|
|
if len(update.BlockIndexes) != len(blocks) {
|
|
t.Errorf("Wrong indexes. Have %d expect %d at %s", len(update.BlockIndexes), len(blocks), caller(1))
|
|
}
|
|
for i := range update.BlockIndexes {
|
|
if update.BlockIndexes[i] != blocks[i] {
|
|
t.Errorf("Index %d incorrect at %s", i, caller(1))
|
|
}
|
|
}
|
|
|
|
if remove {
|
|
fc.downloadProgressMessages = append(fc.downloadProgressMessages[:messageIdx], fc.downloadProgressMessages[messageIdx+1:]...)
|
|
}
|
|
}
|
|
expectEmpty := func() {
|
|
if len(fc.downloadProgressMessages) > 0 {
|
|
t.Errorf("Still have something at %s: %#v", caller(1), fc.downloadProgressMessages)
|
|
}
|
|
}
|
|
|
|
now := time.Now()
|
|
tick := func() time.Time {
|
|
now = now.Add(time.Second)
|
|
return now
|
|
}
|
|
|
|
if len(fc.downloadProgressMessages) != 0 {
|
|
t.Error("Expected no requests")
|
|
}
|
|
|
|
v1 := (protocol.Vector{}).Update(0)
|
|
v2 := (protocol.Vector{}).Update(1)
|
|
|
|
// Requires more than 10 blocks to work.
|
|
blocks := make([]protocol.BlockInfo, 11, 11)
|
|
|
|
state1 := &sharedPullerState{
|
|
folder: "folder",
|
|
file: protocol.FileInfo{
|
|
Name: "state1",
|
|
Version: v1,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
availableUpdated: time.Now(),
|
|
}
|
|
p.registry["1"] = state1
|
|
|
|
// Has no blocks, hence no message is sent
|
|
p.sendDownloadProgressMessages()
|
|
expectEmpty()
|
|
|
|
// Returns update for puller with new extra blocks
|
|
state1.available = []int32{1}
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(0, state1, protocol.UpdateTypeAppend, v1, []int32{1}, true)
|
|
expectEmpty()
|
|
|
|
// Does nothing if nothing changes
|
|
p.sendDownloadProgressMessages()
|
|
expectEmpty()
|
|
|
|
// Does nothing if timestamp updated, but no new blocks (should never happen)
|
|
state1.availableUpdated = tick()
|
|
|
|
p.sendDownloadProgressMessages()
|
|
expectEmpty()
|
|
|
|
// Does not return an update if date blocks change but date does not (should never happen)
|
|
state1.available = []int32{1, 2}
|
|
|
|
p.sendDownloadProgressMessages()
|
|
expectEmpty()
|
|
|
|
// If the date and blocks changes, returns only the diff
|
|
state1.availableUpdated = tick()
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(0, state1, protocol.UpdateTypeAppend, v1, []int32{2}, true)
|
|
expectEmpty()
|
|
|
|
// Returns forget and update if puller version has changed
|
|
state1.file.Version = v2
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(0, state1, protocol.UpdateTypeForget, v1, nil, false)
|
|
expect(1, state1, protocol.UpdateTypeAppend, v2, []int32{1, 2}, true)
|
|
expectEmpty()
|
|
|
|
// Returns forget and append if sharedPullerState creation timer changes.
|
|
|
|
state1.available = []int32{1}
|
|
state1.availableUpdated = tick()
|
|
state1.created = tick()
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(0, state1, protocol.UpdateTypeForget, v2, nil, false)
|
|
expect(1, state1, protocol.UpdateTypeAppend, v2, []int32{1}, true)
|
|
expectEmpty()
|
|
|
|
// Sends an empty update if new file exists, but does not have any blocks yet. (To indicate that the old blocks are no longer available)
|
|
state1.file.Version = v1
|
|
state1.available = nil
|
|
state1.availableUpdated = tick()
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(0, state1, protocol.UpdateTypeForget, v2, nil, false)
|
|
expect(1, state1, protocol.UpdateTypeAppend, v1, nil, true)
|
|
expectEmpty()
|
|
|
|
// Updates for multiple files and folders can be combined
|
|
state1.available = []int32{1, 2, 3}
|
|
state1.availableUpdated = tick()
|
|
|
|
state2 := &sharedPullerState{
|
|
folder: "folder2",
|
|
file: protocol.FileInfo{
|
|
Name: "state2",
|
|
Version: v1,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
state3 := &sharedPullerState{
|
|
folder: "folder",
|
|
file: protocol.FileInfo{
|
|
Name: "state3",
|
|
Version: v1,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
state4 := &sharedPullerState{
|
|
folder: "folder2",
|
|
file: protocol.FileInfo{
|
|
Name: "state4",
|
|
Version: v1,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
p.registry["2"] = state2
|
|
p.registry["3"] = state3
|
|
p.registry["4"] = state4
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(-1, state1, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, false)
|
|
expect(-1, state3, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, true)
|
|
expect(-1, state2, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, false)
|
|
expect(-1, state4, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, true)
|
|
expectEmpty()
|
|
|
|
// Returns forget if puller no longer exists, as well as updates if it has been updated.
|
|
state1.available = []int32{1, 2, 3, 4, 5}
|
|
state1.availableUpdated = tick()
|
|
state2.available = []int32{1, 2, 3, 4, 5}
|
|
state2.availableUpdated = tick()
|
|
|
|
delete(p.registry, "3")
|
|
delete(p.registry, "4")
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(-1, state1, protocol.UpdateTypeAppend, v1, []int32{4, 5}, false)
|
|
expect(-1, state3, protocol.UpdateTypeForget, v1, nil, true)
|
|
expect(-1, state2, protocol.UpdateTypeAppend, v1, []int32{4, 5}, false)
|
|
expect(-1, state4, protocol.UpdateTypeForget, v1, nil, true)
|
|
expectEmpty()
|
|
|
|
// Deletions are sent only once (actual bug I found writing the tests)
|
|
p.sendDownloadProgressMessages()
|
|
p.sendDownloadProgressMessages()
|
|
expectEmpty()
|
|
|
|
// Not sent for "inactive" (symlinks, dirs, or wrong folder) pullers
|
|
// Directory
|
|
state5 := &sharedPullerState{
|
|
folder: "folder",
|
|
file: protocol.FileInfo{
|
|
Name: "state5",
|
|
Version: v1,
|
|
Type: protocol.FileInfoTypeDirectory,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
// Symlink
|
|
state6 := &sharedPullerState{
|
|
folder: "folder",
|
|
file: protocol.FileInfo{
|
|
Name: "state6",
|
|
Version: v1,
|
|
Type: protocol.FileInfoTypeSymlinkUnknown,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
// Some other directory
|
|
state7 := &sharedPullerState{
|
|
folder: "folderXXX",
|
|
file: protocol.FileInfo{
|
|
Name: "state7",
|
|
Version: v1,
|
|
Blocks: blocks,
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
// Less than 10 blocks
|
|
state8 := &sharedPullerState{
|
|
folder: "folder",
|
|
file: protocol.FileInfo{
|
|
Name: "state8",
|
|
Version: v1,
|
|
Blocks: blocks[:3],
|
|
},
|
|
mut: sync.NewRWMutex(),
|
|
available: []int32{1, 2, 3},
|
|
availableUpdated: time.Now(),
|
|
}
|
|
p.registry["5"] = state5
|
|
p.registry["6"] = state6
|
|
p.registry["7"] = state7
|
|
p.registry["8"] = state8
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expectEmpty()
|
|
|
|
// Device is no longer subscribed to a particular folder
|
|
delete(p.registry, "1") // Clean up first
|
|
delete(p.registry, "2") // Clean up first
|
|
|
|
p.sendDownloadProgressMessages()
|
|
expect(-1, state1, protocol.UpdateTypeForget, v1, nil, true)
|
|
expect(-1, state2, protocol.UpdateTypeForget, v1, nil, true)
|
|
|
|
expectEmpty()
|
|
|
|
p.registry["1"] = state1
|
|
p.registry["2"] = state2
|
|
p.registry["3"] = state3
|
|
p.registry["4"] = state4
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
expect(-1, state1, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3, 4, 5}, false)
|
|
expect(-1, state3, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, true)
|
|
expect(-1, state2, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3, 4, 5}, false)
|
|
expect(-1, state4, protocol.UpdateTypeAppend, v1, []int32{1, 2, 3}, true)
|
|
expectEmpty()
|
|
|
|
p.temporaryIndexUnsubscribe(fc)
|
|
p.temporaryIndexSubscribe(fc, []string{"folder"})
|
|
|
|
p.sendDownloadProgressMessages()
|
|
|
|
// See progressemitter.go for explanation why this is commented out.
|
|
// Search for state.cleanup
|
|
//expect(-1, state2, protocol.UpdateTypeForget, v1, nil, false)
|
|
//expect(-1, state4, protocol.UpdateTypeForget, v1, nil, true)
|
|
|
|
expectEmpty()
|
|
|
|
// Cleanup when device no longer exists
|
|
p.temporaryIndexUnsubscribe(fc)
|
|
|
|
p.sendDownloadProgressMessages()
|
|
_, ok := p.sentDownloadStates[fc.ID()]
|
|
if ok {
|
|
t.Error("Should not be there")
|
|
}
|
|
}
|