123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481 |
- // 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 https://mozilla.org/MPL/2.0/.
- package model
- import (
- "context"
- "fmt"
- "os"
- "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) {
- ctx, cancel := context.WithCancel(context.Background())
- evLogger := events.NewLogger()
- go evLogger.Serve(ctx)
- defer cancel()
- w := evLogger.Subscribe(events.DownloadProgress)
- c, cfgCancel := newConfigWrapper(config.Configuration{Version: config.CurrentVersion})
- defer os.Remove(c.ConfigPath())
- defer cfgCancel()
- waiter, err := c.Modify(func(cfg *config.Configuration) {
- cfg.Options.ProgressUpdateIntervalS = 60 // irrelevant, but must be positive
- })
- if err != nil {
- t.Fatal(err)
- }
- waiter.Wait()
- p := NewProgressEmitter(c, evLogger)
- go p.Serve(ctx)
- p.interval = 0
- 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(1)
- 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, cfgCancel := newConfigWrapper(config.Configuration{Version: config.CurrentVersion})
- defer os.Remove(c.ConfigPath())
- defer cfgCancel()
- waiter, err := c.Modify(func(cfg *config.Configuration) {
- cfg.Options.ProgressUpdateIntervalS = 60 // irrelevant, but must be positive
- cfg.Options.TempIndexMinBlocks = 10
- })
- if err != nil {
- t.Fatal(err)
- }
- waiter.Wait()
- fc := newFakeConnection(protocol.DeviceID{}, nil)
- ctx, cancel := context.WithCancel(context.Background())
- evLogger := events.NewLogger()
- go evLogger.Serve(ctx)
- defer cancel()
- p := NewProgressEmitter(c, evLogger)
- p.temporaryIndexSubscribe(fc, []string{"folder", "folder2"})
- p.registry["folder"] = make(map[string]*sharedPullerState)
- p.registry["folder2"] = make(map[string]*sharedPullerState)
- p.registry["folderXXX"] = make(map[string]*sharedPullerState)
- expect := func(updateIdx int, state *sharedPullerState, updateType protocol.FileDownloadProgressUpdateType, version protocol.Vector, blocks []int, 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)
- state1 := &sharedPullerState{
- folder: "folder",
- file: protocol.FileInfo{
- Name: "state1",
- Version: v1,
- Blocks: blocks,
- },
- mut: sync.NewRWMutex(),
- availableUpdated: time.Now(),
- }
- p.registry["folder"]["1"] = state1
- // Has no blocks, hence no message is sent
- sendMsgs(p)
- expectEmpty()
- // Returns update for puller with new extra blocks
- state1.available = []int{1}
- sendMsgs(p)
- expect(0, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1}, true)
- expectEmpty()
- // Does nothing if nothing changes
- sendMsgs(p)
- expectEmpty()
- // Does nothing if timestamp updated, but no new blocks (should never happen)
- state1.availableUpdated = tick()
- sendMsgs(p)
- expectEmpty()
- // Does not return an update if date blocks change but date does not (should never happen)
- state1.available = []int{1, 2}
- sendMsgs(p)
- expectEmpty()
- // If the date and blocks changes, returns only the diff
- state1.availableUpdated = tick()
- sendMsgs(p)
- expect(0, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{2}, true)
- expectEmpty()
- // Returns forget and update if puller version has changed
- state1.file.Version = v2
- sendMsgs(p)
- expect(0, state1, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, false)
- expect(1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v2, []int{1, 2}, true)
- expectEmpty()
- // Returns forget and append if sharedPullerState creation timer changes.
- state1.available = []int{1}
- state1.availableUpdated = tick()
- state1.created = tick()
- sendMsgs(p)
- expect(0, state1, protocol.FileDownloadProgressUpdateTypeForget, v2, nil, false)
- expect(1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v2, []int{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()
- sendMsgs(p)
- expect(0, state1, protocol.FileDownloadProgressUpdateTypeForget, v2, nil, false)
- expect(1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, nil, true)
- expectEmpty()
- // Updates for multiple files and folders can be combined
- state1.available = []int{1, 2, 3}
- state1.availableUpdated = tick()
- state2 := &sharedPullerState{
- folder: "folder2",
- file: protocol.FileInfo{
- Name: "state2",
- Version: v1,
- Blocks: blocks,
- },
- mut: sync.NewRWMutex(),
- available: []int{1, 2, 3},
- availableUpdated: time.Now(),
- }
- state3 := &sharedPullerState{
- folder: "folder",
- file: protocol.FileInfo{
- Name: "state3",
- Version: v1,
- Blocks: blocks,
- },
- mut: sync.NewRWMutex(),
- available: []int{1, 2, 3},
- availableUpdated: time.Now(),
- }
- state4 := &sharedPullerState{
- folder: "folder2",
- file: protocol.FileInfo{
- Name: "state4",
- Version: v1,
- Blocks: blocks,
- },
- mut: sync.NewRWMutex(),
- available: []int{1, 2, 3},
- availableUpdated: time.Now(),
- }
- p.registry["folder2"]["2"] = state2
- p.registry["folder"]["3"] = state3
- p.registry["folder2"]["4"] = state4
- sendMsgs(p)
- expect(-1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, false)
- expect(-1, state3, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, true)
- expect(-1, state2, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, false)
- expect(-1, state4, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, true)
- expectEmpty()
- // Returns forget if puller no longer exists, as well as updates if it has been updated.
- state1.available = []int{1, 2, 3, 4, 5}
- state1.availableUpdated = tick()
- state2.available = []int{1, 2, 3, 4, 5}
- state2.availableUpdated = tick()
- delete(p.registry["folder"], "3")
- delete(p.registry["folder2"], "4")
- sendMsgs(p)
- expect(-1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{4, 5}, false)
- expect(-1, state3, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, true)
- expect(-1, state2, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{4, 5}, false)
- expect(-1, state4, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, true)
- expectEmpty()
- // Deletions are sent only once (actual bug I found writing the tests)
- sendMsgs(p)
- sendMsgs(p)
- 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: []int{1, 2, 3},
- availableUpdated: time.Now(),
- }
- // Symlink
- state6 := &sharedPullerState{
- folder: "folder",
- file: protocol.FileInfo{
- Name: "state6",
- Version: v1,
- Type: protocol.FileInfoTypeSymlink,
- },
- mut: sync.NewRWMutex(),
- available: []int{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: []int{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: []int{1, 2, 3},
- availableUpdated: time.Now(),
- }
- p.registry["folder"]["5"] = state5
- p.registry["folder"]["6"] = state6
- p.registry["folderXXX"]["7"] = state7
- p.registry["folder"]["8"] = state8
- sendMsgs(p)
- expectEmpty()
- // Device is no longer subscribed to a particular folder
- delete(p.registry["folder"], "1") // Clean up first
- delete(p.registry["folder2"], "2") // Clean up first
- sendMsgs(p)
- expect(-1, state1, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, true)
- expect(-1, state2, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, true)
- expectEmpty()
- p.registry["folder"]["1"] = state1
- p.registry["folder2"]["2"] = state2
- p.registry["folder"]["3"] = state3
- p.registry["folder2"]["4"] = state4
- sendMsgs(p)
- expect(-1, state1, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3, 4, 5}, false)
- expect(-1, state3, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, true)
- expect(-1, state2, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3, 4, 5}, false)
- expect(-1, state4, protocol.FileDownloadProgressUpdateTypeAppend, v1, []int{1, 2, 3}, true)
- expectEmpty()
- p.temporaryIndexUnsubscribe(fc)
- p.temporaryIndexSubscribe(fc, []string{"folder"})
- sendMsgs(p)
- // See progressemitter.go for explanation why this is commented out.
- // Search for state.cleanup
- // expect(-1, state2, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, false)
- // expect(-1, state4, protocol.FileDownloadProgressUpdateTypeForget, v1, nil, true)
- expectEmpty()
- // Cleanup when device no longer exists
- p.temporaryIndexUnsubscribe(fc)
- sendMsgs(p)
- _, ok := p.sentDownloadStates[fc.DeviceID()]
- if ok {
- t.Error("Should not be there")
- }
- }
- func sendMsgs(p *ProgressEmitter) {
- p.mut.Lock()
- updates := p.computeProgressUpdates()
- p.mut.Unlock()
- ctx := context.Background()
- for _, update := range updates {
- update.send(ctx)
- }
- }
|