Complete rewrite of the puller (fixes #638, fixes #715, fixes #701)

CONTRIBUTING.md

@@ -39,6 +39,17 @@ attention. In general, any open issues are fair game! Be prepared for a
 review](https://discourse.syncthing.net/t/733); it's all in the name of
 quality. :)
 
+You may be asked to "rebase" or "squash" your pull request before it's
+accepted. This means to make sure that the pull request:
+
+ - Is based on the latest commit on `master`.
+
+ - Only contains a single commit (exceptions to be made where
+   appropriate).
+
+ - Has a descriptive commit message, ending in `(fixes #123)` if it
+   resolves an existing issue.
+
 ## Licensing
 
 All contributions are made under the same MIT License as the rest of the

build.go

@@ -233,11 +233,11 @@ func setBuildEnv() {
 }
 
 func assets() {
-	runPipe("auto/gui.files.go", "godep", "go", "run", "cmd/genassets/main.go", "gui")
+	runPipe("internal/auto/gui.files.go", "godep", "go", "run", "cmd/genassets/main.go", "gui")
 }
 
 func xdr() {
-	for _, f := range []string{"discover/packets", "files/leveldb", "protocol/message"} {
+	for _, f := range []string{"internal/discover/packets", "internal/files/leveldb", "internal/protocol/message"} {
 		runPipe(f+"_xdr.go", "go", "run", "./Godeps/_workspace/src/github.com/calmh/xdr/cmd/genxdr/main.go", "--", f+".go")
 	}
 }

check-contrib.sh

@@ -8,6 +8,8 @@ missing-contribs() {
 
 no-docs-typos() {
 	# Commits that are known to not change code
+	grep -v 63bd0136fb40a91efaa279cb4b4159d82e8e6904 |\
+	grep -v 4e2feb6fbc791bb8a2daf0ab8efb10775d66343e |\
 	grep -v f2459ef3319b2f060dbcdacd0c35a1788a94b8bd |\
 	grep -v b61f418bf2d1f7d5a9d7088a20a2a448e5e66801 |\
 	grep -v f0621207e3953711f9ab86d99724f1d0faac45b1 |\

cmd/stindex/main.go

@@ -10,8 +10,8 @@ import (
 	"log"
 	"os"
 
-	"github.com/syncthing/syncthing/files"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/files"
+	"github.com/syncthing/syncthing/internal/protocol"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 

cmd/syncthing/gui.go

@@ -21,13 +21,13 @@ import (
 	"time"
 
 	"code.google.com/p/go.crypto/bcrypt"
-	"github.com/syncthing/syncthing/auto"
-	"github.com/syncthing/syncthing/config"
-	"github.com/syncthing/syncthing/events"
-	"github.com/syncthing/syncthing/logger"
-	"github.com/syncthing/syncthing/model"
-	"github.com/syncthing/syncthing/protocol"
-	"github.com/syncthing/syncthing/upgrade"
+	"github.com/syncthing/syncthing/internal/auto"
+	"github.com/syncthing/syncthing/internal/config"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/logger"
+	"github.com/syncthing/syncthing/internal/model"
+	"github.com/syncthing/syncthing/internal/protocol"
+	"github.com/syncthing/syncthing/internal/upgrade"
 	"github.com/vitrun/qart/qr"
 )
 
@@ -87,7 +87,6 @@ func startGUI(cfg config.GUIConfiguration, assetDir string, m *model.Model) erro
 	getRestMux.HandleFunc("/rest/ignores", withModel(m, restGetIgnores))
 	getRestMux.HandleFunc("/rest/lang", restGetLang)
 	getRestMux.HandleFunc("/rest/model", withModel(m, restGetModel))
-	getRestMux.HandleFunc("/rest/model/version", withModel(m, restGetModelVersion))
 	getRestMux.HandleFunc("/rest/need", withModel(m, restGetNeed))
 	getRestMux.HandleFunc("/rest/nodeid", restGetNodeID)
 	getRestMux.HandleFunc("/rest/report", withModel(m, restGetReport))
@@ -238,17 +237,6 @@ func restGetCompletion(m *model.Model, w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(res)
 }
 
-func restGetModelVersion(m *model.Model, w http.ResponseWriter, r *http.Request) {
-	var qs = r.URL.Query()
-	var repo = qs.Get("repo")
-	var res = make(map[string]interface{})
-
-	res["version"] = m.LocalVersion(repo)
-
-	w.Header().Set("Content-Type", "application/json; charset=utf-8")
-	json.NewEncoder(w).Encode(res)
-}
-
 func restGetModel(m *model.Model, w http.ResponseWriter, r *http.Request) {
 	var qs = r.URL.Query()
 	var repo = qs.Get("repo")
@@ -273,7 +261,7 @@ func restGetModel(m *model.Model, w http.ResponseWriter, r *http.Request) {
 	res["inSyncFiles"], res["inSyncBytes"] = globalFiles-needFiles, globalBytes-needBytes
 
 	res["state"], res["stateChanged"] = m.State(repo)
-	res["version"] = m.LocalVersion(repo)
+	res["version"] = m.CurrentLocalVersion(repo) + m.RemoteLocalVersion(repo)
 
 	w.Header().Set("Content-Type", "application/json; charset=utf-8")
 	json.NewEncoder(w).Encode(res)

cmd/syncthing/gui_auth.go

@@ -14,7 +14,7 @@ import (
 	"time"
 
 	"code.google.com/p/go.crypto/bcrypt"
-	"github.com/syncthing/syncthing/config"
+	"github.com/syncthing/syncthing/internal/config"
 )
 
 var (

cmd/syncthing/gui_csrf.go

@@ -16,7 +16,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/syncthing/syncthing/osutil"
+	"github.com/syncthing/syncthing/internal/osutil"
 )
 
 var csrfTokens []string

cmd/syncthing/main.go

@@ -27,15 +27,15 @@ import (
 
 	"code.google.com/p/go.crypto/bcrypt"
 	"github.com/juju/ratelimit"
-	"github.com/syncthing/syncthing/config"
-	"github.com/syncthing/syncthing/discover"
-	"github.com/syncthing/syncthing/events"
-	"github.com/syncthing/syncthing/files"
-	"github.com/syncthing/syncthing/logger"
-	"github.com/syncthing/syncthing/model"
-	"github.com/syncthing/syncthing/protocol"
-	"github.com/syncthing/syncthing/upgrade"
-	"github.com/syncthing/syncthing/upnp"
+	"github.com/syncthing/syncthing/internal/config"
+	"github.com/syncthing/syncthing/internal/discover"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/files"
+	"github.com/syncthing/syncthing/internal/logger"
+	"github.com/syncthing/syncthing/internal/model"
+	"github.com/syncthing/syncthing/internal/protocol"
+	"github.com/syncthing/syncthing/internal/upgrade"
+	"github.com/syncthing/syncthing/internal/upnp"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/opt"
 )
@@ -442,7 +442,7 @@ nextRepo:
 		m.AddRepo(repo)
 
 		fi, err := os.Stat(repo.Directory)
-		if m.LocalVersion(repo.ID) > 0 {
+		if m.CurrentLocalVersion(repo.ID) > 0 {
 			// Safety check. If the cached index contains files but the
 			// repository doesn't exist, we have a problem. We would assume
 			// that all files have been deleted which might not be the case,
@@ -453,8 +453,8 @@ nextRepo:
 				continue nextRepo
 			}
 		} else if os.IsNotExist(err) {
-			// If we don't have ny files in the index, and the directory does
-			// exist, try creating it.
+			// If we don't have any files in the index, and the directory
+			// doesn't exist, try creating it.
 			err = os.MkdirAll(repo.Directory, 0700)
 		}
 
@@ -582,7 +582,7 @@ nextRepo:
 			m.StartRepoRO(repo.ID)
 		} else {
 			l.Okf("Ready to synchronize %s (read-write)", repo.ID)
-			m.StartRepoRW(repo.ID, cfg.Options.ParallelRequests)
+			m.StartRepoRW(repo.ID)
 		}
 	}
 
@@ -1159,7 +1159,7 @@ func standbyMonitor() {
 	for {
 		time.Sleep(10 * time.Second)
 		if time.Since(now) > 2*time.Minute {
-			l.Infoln("Paused state detected, possibly woke up from standby. Restarting in", restartDelay)
+			l.Infof("Paused state detected, possibly woke up from standby. Restarting in %v.", restartDelay)
 
 			// We most likely just woke from standby. If we restart
 			// immediately chances are we won't have networking ready. Give

cmd/syncthing/perfstats_unix.go

@@ -12,6 +12,8 @@ import (
 	"runtime"
 	"syscall"
 	"time"
+
+	"github.com/syncthing/syncthing/internal/protocol"
 )
 
 func init() {
@@ -30,6 +32,7 @@ func savePerfStats(file string) {
 	var prevTime int64
 	var rusage syscall.Rusage
 	var memstats runtime.MemStats
+	var prevIn, prevOut uint64
 
 	t0 := time.Now()
 	for t := range time.NewTicker(250 * time.Millisecond).C {
@@ -41,11 +44,18 @@ func savePerfStats(file string) {
 		cpuUsagePercent := 100 * float64(usageDiff) / float64(timeDiff)
 		prevTime = curTime
 		prevUsage = curUsage
+		in, out := protocol.TotalInOut()
+		var inRate, outRate float64
+		if timeDiff > 0 {
+			inRate = float64(in-prevIn) / (float64(timeDiff) / 1e9)    // bytes per second
+			outRate = float64(out-prevOut) / (float64(timeDiff) / 1e9) // bytes per second
+		}
+		prevIn, prevOut = in, out
 
 		runtime.ReadMemStats(&memstats)
 
 		startms := int(t.Sub(t0).Seconds() * 1000)
 
-		fmt.Fprintf(fd, "%d\t%f\t%d\t%d\n", startms, cpuUsagePercent, memstats.Alloc, memstats.Sys-memstats.HeapReleased)
+		fmt.Fprintf(fd, "%d\t%f\t%d\t%d\t%.0f\t%.0f\n", startms, cpuUsagePercent, memstats.Alloc, memstats.Sys-memstats.HeapReleased, inRate, outRate)
 	}
 }

cmd/syncthing/usage_report.go

@@ -15,7 +15,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/syncthing/syncthing/model"
+	"github.com/syncthing/syncthing/internal/model"
 )
 
 // Current version number of the usage report, for acceptance purposes. If

gui/index.html

@@ -164,8 +164,10 @@
             </div>
           </div>
         </div>
-        <button class="btn btn-sm btn-default pull-right" ng-click="addRepo()"><span class="glyphicon glyphicon-plus"></span>&emsp;<span translate>Add Repository</span></button>
-        <div class="clearfix"></div>
+        <div class="form-group">
+          <button class="btn btn-sm btn-default pull-right" ng-click="addRepo()"><span class="glyphicon glyphicon-plus"></span>&emsp;<span translate>Add Repository</span></button>
+          <div class="clearfix"></div>
+        </div>
         <hr class="visible-sm"/>
       </div>
 
@@ -284,8 +286,10 @@
             </div>
           </div>
         </div>
-        <button class="btn btn-sm btn-default pull-right" ng-click="addNode()"><span class="glyphicon glyphicon-plus"></span>&emsp;<span translate>Add Node</span></button>
-        <div class="clearfix"></div>
+        <div class="form-group">
+          <button class="btn btn-sm btn-default pull-right" ng-click="addNode()"><span class="glyphicon glyphicon-plus"></span>&emsp;<span translate>Add Node</span></button>
+          <div class="clearfix"></div>
+        </div>
       </div>
     </div> <!-- /row -->
 

internal/auto/auto_test.go

@@ -8,7 +8,7 @@ import (
 	"bytes"
 	"testing"
 
-	"github.com/syncthing/syncthing/auto"
+	"github.com/syncthing/syncthing/internal/auto"
 )
 
 func TestAssets(t *testing.T) {

internal/beacon/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/config/config.go

@@ -14,10 +14,10 @@ import (
 	"strconv"
 
 	"code.google.com/p/go.crypto/bcrypt"
-	"github.com/syncthing/syncthing/events"
-	"github.com/syncthing/syncthing/logger"
-	"github.com/syncthing/syncthing/osutil"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/logger"
+	"github.com/syncthing/syncthing/internal/osutil"
+	"github.com/syncthing/syncthing/internal/protocol"
 )
 
 var l = logger.DefaultLogger
@@ -118,7 +118,6 @@ type OptionsConfiguration struct {
 	LocalAnnEnabled    bool     `xml:"localAnnounceEnabled" default:"true"`
 	LocalAnnPort       int      `xml:"localAnnouncePort" default:"21025"`
 	LocalAnnMCAddr     string   `xml:"localAnnounceMCAddr" default:"[ff32::5222]:21026"`
-	ParallelRequests   int      `xml:"parallelRequests" default:"16"`
 	MaxSendKbps        int      `xml:"maxSendKbps"`
 	MaxRecvKbps        int      `xml:"maxRecvKbps"`
 	ReconnectIntervalS int      `xml:"reconnectionIntervalS" default:"60"`

internal/config/config_test.go

@@ -9,7 +9,7 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/protocol"
 )
 
 var node1, node2, node3, node4 protocol.NodeID
@@ -29,7 +29,6 @@ func TestDefaultValues(t *testing.T) {
 		LocalAnnEnabled:    true,
 		LocalAnnPort:       21025,
 		LocalAnnMCAddr:     "[ff32::5222]:21026",
-		ParallelRequests:   16,
 		MaxSendKbps:        0,
 		MaxRecvKbps:        0,
 		ReconnectIntervalS: 60,
@@ -121,7 +120,6 @@ func TestOverriddenValues(t *testing.T) {
 		LocalAnnEnabled:    false,
 		LocalAnnPort:       42123,
 		LocalAnnMCAddr:     "quux:3232",
-		ParallelRequests:   32,
 		MaxSendKbps:        1234,
 		MaxRecvKbps:        2341,
 		ReconnectIntervalS: 6000,

internal/discover/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/discover/discover.go

@@ -14,9 +14,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/syncthing/syncthing/beacon"
-	"github.com/syncthing/syncthing/events"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/beacon"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/protocol"
 )
 
 type Discoverer struct {

internal/events/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/events/events_test.go

@@ -9,7 +9,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/syncthing/syncthing/events"
+	"github.com/syncthing/syncthing/internal/events"
 )
 
 var timeout = 100 * time.Millisecond

internal/files/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/files/leveldb.go

@@ -10,8 +10,8 @@ import (
 	"sort"
 	"sync"
 
-	"github.com/syncthing/syncthing/lamport"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/lamport"
+	"github.com/syncthing/syncthing/internal/protocol"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
 	"github.com/syndtr/goleveldb/leveldb/opt"
@@ -77,8 +77,8 @@ keyTypeNode (1 byte)
     repository (64 bytes)
         node (32 bytes)
             name (variable size)
-            	|
-            	scanner.File
+		|
+		scanner.File
 
 keyTypeGlobal (1 byte)
 	repository (64 bytes)

internal/files/set.go

@@ -13,8 +13,8 @@ package files
 import (
 	"sync"
 
-	"github.com/syncthing/syncthing/lamport"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/lamport"
+	"github.com/syncthing/syncthing/internal/protocol"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 

internal/files/set_test.go

@@ -11,9 +11,9 @@ import (
 	"sort"
 	"testing"
 
-	"github.com/syncthing/syncthing/files"
-	"github.com/syncthing/syncthing/lamport"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/files"
+	"github.com/syncthing/syncthing/internal/lamport"
+	"github.com/syncthing/syncthing/internal/protocol"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 )

internal/ignore/ignore.go

@@ -13,7 +13,7 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/syncthing/syncthing/fnmatch"
+	"github.com/syncthing/syncthing/internal/fnmatch"
 )
 
 type Pattern struct {

internal/ignore/ignore_test.go

@@ -10,7 +10,7 @@ import (
 	"runtime"
 	"testing"
 
-	"github.com/syncthing/syncthing/ignore"
+	"github.com/syncthing/syncthing/internal/ignore"
 )
 
 func TestIgnore(t *testing.T) {

internal/luhn/luhn_test.go

@@ -7,7 +7,7 @@ package luhn_test
 import (
 	"testing"
 
-	"github.com/syncthing/syncthing/luhn"
+	"github.com/syncthing/syncthing/internal/luhn"
 )
 
 func TestGenerate(t *testing.T) {

internal/model/blockqueue.go

@@ -4,7 +4,7 @@
 
 package model
 
-import "github.com/syncthing/syncthing/protocol"
+import "github.com/syncthing/syncthing/internal/protocol"
 
 type bqAdd struct {
 	file protocol.FileInfo

internal/model/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/model/model.go

@@ -19,15 +19,16 @@ import (
 	"sync"
 	"time"
 
-	"github.com/syncthing/syncthing/config"
-	"github.com/syncthing/syncthing/events"
-	"github.com/syncthing/syncthing/files"
-	"github.com/syncthing/syncthing/ignore"
-	"github.com/syncthing/syncthing/lamport"
-	"github.com/syncthing/syncthing/osutil"
-	"github.com/syncthing/syncthing/protocol"
-	"github.com/syncthing/syncthing/scanner"
-	"github.com/syncthing/syncthing/stats"
+	"github.com/syncthing/syncthing/internal/config"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/files"
+	"github.com/syncthing/syncthing/internal/ignore"
+	"github.com/syncthing/syncthing/internal/lamport"
+	"github.com/syncthing/syncthing/internal/osutil"
+	"github.com/syncthing/syncthing/internal/protocol"
+	"github.com/syncthing/syncthing/internal/scanner"
+	"github.com/syncthing/syncthing/internal/stats"
+	"github.com/syncthing/syncthing/internal/versioner"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 
@@ -138,22 +139,54 @@ func NewModel(indexDir string, cfg *config.Configuration, nodeName, clientName,
 // StartRW starts read/write processing on the current model. When in
 // read/write mode the model will attempt to keep in sync with the cluster by
 // pulling needed files from peer nodes.
-func (m *Model) StartRepoRW(repo string, threads int) {
-	m.rmut.RLock()
-	defer m.rmut.RUnlock()
+func (m *Model) StartRepoRW(repo string) {
+	m.rmut.Lock()
+	cfg, ok := m.repoCfgs[repo]
+	m.rmut.Unlock()
 
-	if cfg, ok := m.repoCfgs[repo]; !ok {
-		panic("cannot start without repo")
-	} else {
-		newPuller(cfg, m, threads, m.cfg)
+	if !ok {
+		panic("cannot start nonexistent repo " + repo)
 	}
+
+	p := Puller{
+		repo:     repo,
+		dir:      cfg.Directory,
+		scanIntv: time.Duration(cfg.RescanIntervalS) * time.Second,
+		model:    m,
+	}
+
+	if len(cfg.Versioning.Type) > 0 {
+		factory, ok := versioner.Factories[cfg.Versioning.Type]
+		if !ok {
+			l.Fatalf("Requested versioning type %q that does not exist", cfg.Versioning.Type)
+		}
+		p.versioner = factory(repo, cfg.Directory, cfg.Versioning.Params)
+	}
+
+	go p.Serve()
 }
 
 // StartRO starts read only processing on the current model. When in
 // read only mode the model will announce files to the cluster but not
 // pull in any external changes.
 func (m *Model) StartRepoRO(repo string) {
-	m.StartRepoRW(repo, 0) // zero threads => read only
+	intv := time.Duration(m.repoCfgs[repo].RescanIntervalS) * time.Second
+	go func() {
+		for {
+			time.Sleep(intv)
+
+			if debug {
+				l.Debugln(m, "rescan", repo)
+			}
+
+			m.setState(repo, RepoScanning)
+			if err := m.ScanRepo(repo); err != nil {
+				invalidateRepo(m.cfg, repo, err)
+				return
+			}
+			m.setState(repo, RepoIdle)
+		}
+	}()
 }
 
 type ConnectionInfo struct {
@@ -240,7 +273,7 @@ func (m *Model) Completion(node protocol.NodeID, repo string) float64 {
 
 	res := 100 * (1 - float64(need)/float64(tot))
 	if debug {
-		l.Debugf("Completion(%s, %q): %f (%d / %d)", node, repo, res, need, tot)
+		l.Debugf("%v Completion(%s, %q): %f (%d / %d)", m, node, repo, res, need, tot)
 	}
 
 	return res
@@ -316,7 +349,7 @@ func (m *Model) NeedSize(repo string) (files int, bytes int64) {
 		})
 	}
 	if debug {
-		l.Debugf("NeedSize(%q): %d %d", repo, files, bytes)
+		l.Debugf("%v NeedSize(%q): %d %d", m, repo, files, bytes)
 	}
 	return
 }
@@ -389,7 +422,7 @@ func (m *Model) Index(nodeID protocol.NodeID, repo string, fs []protocol.FileInf
 // Implements the protocol.Model interface.
 func (m *Model) IndexUpdate(nodeID protocol.NodeID, repo string, fs []protocol.FileInfo) {
 	if debug {
-		l.Debugf("IDXUP(in): %s / %q: %d files", nodeID, repo, len(fs))
+		l.Debugf("%v IDXUP(in): %s / %q: %d files", m, nodeID, repo, len(fs))
 	}
 
 	if !m.repoSharedWith(repo, nodeID) {
@@ -449,7 +482,7 @@ func (m *Model) ClusterConfig(nodeID protocol.NodeID, cm protocol.ClusterConfigM
 	l.Infof(`Node %s client is "%s %s"`, nodeID, cm.ClientName, cm.ClientVersion)
 
 	if name := cm.GetOption("name"); name != "" {
-		l.Infof("Node %s hostname is %q", nodeID, name)
+		l.Infof("Node %s name is %q", nodeID, name)
 		node := m.cfg.GetNodeConfiguration(nodeID)
 		if node != nil && node.Name == "" {
 			node.Name = name
@@ -475,7 +508,7 @@ func (m *Model) ClusterConfig(nodeID protocol.NodeID, cm protocol.ClusterConfigM
 				var id protocol.NodeID
 				copy(id[:], node.ID)
 
-				if _, ok := m.nodeRepos[id]; !ok {
+				if m.cfg.GetNodeConfiguration(id) == nil {
 					// The node is currently unknown. Add it to the config.
 
 					l.Infof("Adding node %v to config (vouched for by introducer %v)", id, nodeID)
@@ -574,20 +607,20 @@ func (m *Model) Request(nodeID protocol.NodeID, repo, name string, offset int64,
 	lf := r.Get(protocol.LocalNodeID, name)
 	if protocol.IsInvalid(lf.Flags) || protocol.IsDeleted(lf.Flags) {
 		if debug {
-			l.Debugf("REQ(in): %s: %q / %q o=%d s=%d; invalid: %v", nodeID, repo, name, offset, size, lf)
+			l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d; invalid: %v", m, nodeID, repo, name, offset, size, lf)
 		}
 		return nil, ErrInvalid
 	}
 
 	if offset > lf.Size() {
 		if debug {
-			l.Debugf("REQ(in; nonexistent): %s: %q o=%d s=%d", nodeID, name, offset, size)
+			l.Debugf("%v REQ(in; nonexistent): %s: %q o=%d s=%d", m, nodeID, name, offset, size)
 		}
 		return nil, ErrNoSuchFile
 	}
 
 	if debug && nodeID != protocol.LocalNodeID {
-		l.Debugf("REQ(in): %s: %q / %q o=%d s=%d", nodeID, repo, name, offset, size)
+		l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d", m, nodeID, repo, name, offset, size)
 	}
 	m.rmut.RLock()
 	fn := filepath.Join(m.repoCfgs[repo].Directory, name)
@@ -768,15 +801,9 @@ func sendIndexes(conn protocol.Connection, repo string, fs *files.Set, ignores i
 	var err error
 
 	if debug {
-		l.Debugf("sendIndexes for %s-%s@/%q starting", nodeID, name, repo)
+		l.Debugf("sendIndexes for %s-%s/%q starting", nodeID, name, repo)
 	}
 
-	defer func() {
-		if debug {
-			l.Debugf("sendIndexes for %s-%s@/%q exiting: %v", nodeID, name, repo, err)
-		}
-	}()
-
 	minLocalVer, err := sendIndexTo(true, 0, conn, repo, fs, ignores)
 
 	for err == nil {
@@ -787,6 +814,10 @@ func sendIndexes(conn protocol.Connection, repo string, fs *files.Set, ignores i
 
 		minLocalVer, err = sendIndexTo(false, minLocalVer, conn, repo, fs, ignores)
 	}
+
+	if debug {
+		l.Debugf("sendIndexes for %s-%s/%q exiting: %v", nodeID, name, repo, err)
+	}
 }
 
 func sendIndexTo(initial bool, minLocalVer uint64, conn protocol.Connection, repo string, fs *files.Set, ignores ignore.Patterns) (uint64, error) {
@@ -877,7 +908,7 @@ func (m *Model) requestGlobal(nodeID protocol.NodeID, repo, name string, offset
 	}
 
 	if debug {
-		l.Debugf("REQ(out): %s: %q / %q o=%d s=%d h=%x", nodeID, repo, name, offset, size, hash)
+		l.Debugf("%v REQ(out): %s: %q / %q o=%d s=%d h=%x", m, nodeID, repo, name, offset, size, hash)
 	}
 
 	return nc.Request(repo, name, offset, size)
@@ -1175,10 +1206,10 @@ func (m *Model) Override(repo string) {
 	m.setState(repo, RepoIdle)
 }
 
-// Version returns the change version for the given repository. This is
-// guaranteed to increment if the contents of the local or global repository
-// has changed.
-func (m *Model) LocalVersion(repo string) uint64 {
+// CurrentLocalVersion returns the change version for the given repository.
+// This is guaranteed to increment if the contents of the local repository has
+// changed.
+func (m *Model) CurrentLocalVersion(repo string) uint64 {
 	m.rmut.Lock()
 	defer m.rmut.Unlock()
 
@@ -1187,10 +1218,41 @@ func (m *Model) LocalVersion(repo string) uint64 {
 		panic("bug: LocalVersion called for nonexistent repo " + repo)
 	}
 
-	ver := fs.LocalVersion(protocol.LocalNodeID)
+	return fs.LocalVersion(protocol.LocalNodeID)
+}
+
+// RemoteLocalVersion returns the change version for the given repository, as
+// sent by remote peers. This is guaranteed to increment if the contents of
+// the remote or global repository has changed.
+func (m *Model) RemoteLocalVersion(repo string) uint64 {
+	m.rmut.Lock()
+	defer m.rmut.Unlock()
+
+	fs, ok := m.repoFiles[repo]
+	if !ok {
+		panic("bug: LocalVersion called for nonexistent repo " + repo)
+	}
+
+	var ver uint64
 	for _, n := range m.repoNodes[repo] {
 		ver += fs.LocalVersion(n)
 	}
 
 	return ver
 }
+
+func (m *Model) availability(repo string, file string) []protocol.NodeID {
+	m.rmut.Lock()
+	defer m.rmut.Unlock()
+
+	fs, ok := m.repoFiles[repo]
+	if !ok {
+		return nil
+	}
+
+	return fs.Availability(file)
+}
+
+func (m *Model) String() string {
+	return fmt.Sprintf("model@%p", m)
+}

internal/model/model_test.go

@@ -11,8 +11,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/syncthing/syncthing/config"
-	"github.com/syncthing/syncthing/protocol"
+	"github.com/syncthing/syncthing/internal/config"
+	"github.com/syncthing/syncthing/internal/protocol"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 )
@@ -241,25 +241,6 @@ func BenchmarkRequest(b *testing.B) {
 	}
 }
 
-func TestActivityMap(t *testing.T) {
-	isValid := func(protocol.NodeID) bool {
-		return true
-	}
-	m := make(activityMap)
-	if node := m.leastBusyNode([]protocol.NodeID{node1}, isValid); node != node1 {
-		t.Errorf("Incorrect least busy node %q", node)
-	}
-	if node := m.leastBusyNode([]protocol.NodeID{node2}, isValid); node != node2 {
-		t.Errorf("Incorrect least busy node %q", node)
-	}
-	if node := m.leastBusyNode([]protocol.NodeID{node1, node2}, isValid); node != node1 {
-		t.Errorf("Incorrect least busy node %q", node)
-	}
-	if node := m.leastBusyNode([]protocol.NodeID{node1, node2}, isValid); node != node2 {
-		t.Errorf("Incorrect least busy node %q", node)
-	}
-}
-
 func TestNodeRename(t *testing.T) {
 	ccm := protocol.ClusterConfigMessage{
 		ClientName:    "syncthing",

internal/model/nodeactivity.go

@@ -0,0 +1,51 @@
+// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
+// All rights reserved. Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file.
+
+package model
+
+import (
+	"sync"
+
+	"github.com/syncthing/syncthing/internal/protocol"
+)
+
+// nodeActivity tracks the number of outstanding requests per node and can
+// answer which node is least busy. It is safe for use from multiple
+// goroutines.
+type nodeActivity struct {
+	act map[protocol.NodeID]int
+	mut sync.Mutex
+}
+
+func newNodeActivity() *nodeActivity {
+	return &nodeActivity{
+		act: make(map[protocol.NodeID]int),
+	}
+}
+
+func (m nodeActivity) leastBusy(availability []protocol.NodeID) protocol.NodeID {
+	m.mut.Lock()
+	var low int = 2<<30 - 1
+	var selected protocol.NodeID
+	for _, node := range availability {
+		if usage := m.act[node]; usage < low {
+			low = usage
+			selected = node
+		}
+	}
+	m.mut.Unlock()
+	return selected
+}
+
+func (m nodeActivity) using(node protocol.NodeID) {
+	m.mut.Lock()
+	defer m.mut.Unlock()
+	m.act[node]++
+}
+
+func (m nodeActivity) done(node protocol.NodeID) {
+	m.mut.Lock()
+	defer m.mut.Unlock()
+	m.act[node]--
+}

internal/model/nodeactivity_test.go

@@ -0,0 +1,56 @@
+// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
+// All rights reserved. Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file.
+
+package model
+
+import (
+	"testing"
+
+	"github.com/syncthing/syncthing/internal/protocol"
+)
+
+func TestNodeActivity(t *testing.T) {
+	n0 := protocol.NodeID{1, 2, 3, 4}
+	n1 := protocol.NodeID{5, 6, 7, 8}
+	n2 := protocol.NodeID{9, 10, 11, 12}
+	nodes := []protocol.NodeID{n0, n1, n2}
+	na := newNodeActivity()
+
+	if lb := na.leastBusy(nodes); lb != n0 {
+		t.Errorf("Least busy node should be n0 (%v) not %v", n0, lb)
+	}
+	if lb := na.leastBusy(nodes); lb != n0 {
+		t.Errorf("Least busy node should still be n0 (%v) not %v", n0, lb)
+	}
+
+	na.using(na.leastBusy(nodes))
+	if lb := na.leastBusy(nodes); lb != n1 {
+		t.Errorf("Least busy node should be n1 (%v) not %v", n1, lb)
+	}
+
+	na.using(na.leastBusy(nodes))
+	if lb := na.leastBusy(nodes); lb != n2 {
+		t.Errorf("Least busy node should be n2 (%v) not %v", n2, lb)
+	}
+
+	na.using(na.leastBusy(nodes))
+	if lb := na.leastBusy(nodes); lb != n0 {
+		t.Errorf("Least busy node should be n0 (%v) not %v", n0, lb)
+	}
+
+	na.done(n1)
+	if lb := na.leastBusy(nodes); lb != n1 {
+		t.Errorf("Least busy node should be n1 (%v) not %v", n1, lb)
+	}
+
+	na.done(n2)
+	if lb := na.leastBusy(nodes); lb != n1 {
+		t.Errorf("Least busy node should still be n1 (%v) not %v", n1, lb)
+	}
+
+	na.done(n0)
+	if lb := na.leastBusy(nodes); lb != n0 {
+		t.Errorf("Least busy node should be n0 (%v) not %v", n0, lb)
+	}
+}

internal/model/puller.go

@@ -0,0 +1,588 @@
+// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
+// All rights reserved. Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file.
+
+package model
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/syncthing/syncthing/internal/config"
+	"github.com/syncthing/syncthing/internal/events"
+	"github.com/syncthing/syncthing/internal/osutil"
+	"github.com/syncthing/syncthing/internal/protocol"
+	"github.com/syncthing/syncthing/internal/scanner"
+	"github.com/syncthing/syncthing/internal/versioner"
+)
+
+// TODO: Stop on errors
+
+const (
+	copiersPerRepo   = 1
+	pullersPerRepo   = 16
+	finishersPerRepo = 2
+	pauseIntv        = 60 * time.Second
+	nextPullIntv     = 10 * time.Second
+	checkPullIntv    = 1 * time.Second
+)
+
+// A pullBlockState is passed to the puller routine for each block that needs
+// to be fetched.
+type pullBlockState struct {
+	*sharedPullerState
+	block protocol.BlockInfo
+}
+
+// A copyBlocksState is passed to copy routine if the file has blocks to be
+// copied from the original.
+type copyBlocksState struct {
+	*sharedPullerState
+	blocks []protocol.BlockInfo
+}
+
+var (
+	activity  = newNodeActivity()
+	errNoNode = errors.New("no available source node")
+)
+
+type Puller struct {
+	repo      string
+	dir       string
+	scanIntv  time.Duration
+	model     *Model
+	stop      chan struct{}
+	versioner versioner.Versioner
+}
+
+// Serve will run scans and pulls. It will return when Stop()ed or on a
+// critical error.
+func (p *Puller) Serve() {
+	if debug {
+		l.Debugln(p, "starting")
+		defer l.Debugln(p, "exiting")
+	}
+
+	p.stop = make(chan struct{})
+
+	pullTimer := time.NewTimer(checkPullIntv)
+	scanTimer := time.NewTimer(p.scanIntv)
+
+	defer func() {
+		pullTimer.Stop()
+		scanTimer.Stop()
+		// TODO: Should there be an actual RepoStopped state?
+		p.model.setState(p.repo, RepoIdle)
+	}()
+
+	var prevVer uint64
+
+	// Clean out old temporaries before we start pulling
+	p.clean()
+
+loop:
+	for {
+		select {
+		case <-p.stop:
+			return
+
+		// TODO: We could easily add a channel here for notifications from
+		// Index(), so that we immediately start a pull when new index
+		// information is available. Before that though, I'd like to build a
+		// repeatable benchmark of how long it takes to sync a change from
+		// node A to node B, so we have something to work against.
+		case <-pullTimer.C:
+			// RemoteLocalVersion() is a fast call, doesn't touch the database.
+			curVer := p.model.RemoteLocalVersion(p.repo)
+			if curVer == prevVer {
+				pullTimer.Reset(checkPullIntv)
+				continue
+			}
+
+			if debug {
+				l.Debugln(p, "pulling", prevVer, curVer)
+			}
+			p.model.setState(p.repo, RepoSyncing)
+			tries := 0
+			for {
+				tries++
+				changed := p.pullerIteration(copiersPerRepo, pullersPerRepo, finishersPerRepo)
+				if debug {
+					l.Debugln(p, "changed", changed)
+				}
+
+				if changed == 0 {
+					// No files were changed by the puller, so we are in
+					// sync. Remember the local version number and
+					// schedule a resync a little bit into the future.
+					prevVer = curVer
+					pullTimer.Reset(nextPullIntv)
+					break
+				}
+
+				if tries > 10 {
+					// We've tried a bunch of times to get in sync, but
+					// we're not making it. Probably there are write
+					// errors preventing us. Flag this with a warning and
+					// wait a bit longer before retrying.
+					l.Warnf("Repo %q isn't making progress - check logs for possible root cause. Pausing puller for %v.", p.repo, pauseIntv)
+					pullTimer.Reset(pauseIntv)
+					break
+				}
+			}
+			p.model.setState(p.repo, RepoIdle)
+
+		// The reason for running the scanner from within the puller is that
+		// this is the easiest way to make sure we are not doing both at the
+		// same time.
+		case <-scanTimer.C:
+			if debug {
+				l.Debugln(p, "rescan")
+			}
+			p.model.setState(p.repo, RepoScanning)
+			if err := p.model.ScanRepo(p.repo); err != nil {
+				invalidateRepo(p.model.cfg, p.repo, err)
+				break loop
+			}
+			p.model.setState(p.repo, RepoIdle)
+			scanTimer.Reset(p.scanIntv)
+		}
+	}
+}
+
+func (p *Puller) Stop() {
+	close(p.stop)
+}
+
+func (p *Puller) String() string {
+	return fmt.Sprintf("puller/%s@%p", p.repo, p)
+}
+
+// pullerIteration runs a single puller iteration for the given repo and
+// returns the number items that should have been synced (even those that
+// might have failed). One puller iteration handles all files currently
+// flagged as needed in the repo. The specified number of copier, puller and
+// finisher routines are used. It's seldom efficient to use more than one
+// copier routine, while multiple pullers are essential and multiple finishers
+// may be useful (they are primarily CPU bound due to hashing).
+func (p *Puller) pullerIteration(ncopiers, npullers, nfinishers int) int {
+	pullChan := make(chan pullBlockState)
+	copyChan := make(chan copyBlocksState)
+	finisherChan := make(chan *sharedPullerState)
+
+	var wg sync.WaitGroup
+	var doneWg sync.WaitGroup
+
+	for i := 0; i < ncopiers; i++ {
+		wg.Add(1)
+		go func() {
+			// copierRoutine finishes when copyChan is closed
+			p.copierRoutine(copyChan, finisherChan)
+			wg.Done()
+		}()
+	}
+
+	for i := 0; i < npullers; i++ {
+		wg.Add(1)
+		go func() {
+			// pullerRoutine finishes when pullChan is closed
+			p.pullerRoutine(pullChan, finisherChan)
+			wg.Done()
+		}()
+	}
+
+	for i := 0; i < nfinishers; i++ {
+		doneWg.Add(1)
+		// finisherRoutine finishes when finisherChan is closed
+		go func() {
+			p.finisherRoutine(finisherChan)
+			doneWg.Done()
+		}()
+	}
+
+	p.model.rmut.RLock()
+	files := p.model.repoFiles[p.repo]
+	p.model.rmut.RUnlock()
+
+	// !!!
+	// WithNeed takes a database snapshot (by necessity). By the time we've
+	// handled a bunch of files it might have become out of date and we might
+	// be attempting to sync with an old version of a file...
+	// !!!
+
+	changed := 0
+	files.WithNeed(protocol.LocalNodeID, func(intf protocol.FileIntf) bool {
+		file := intf.(protocol.FileInfo)
+
+		events.Default.Log(events.ItemStarted, map[string]string{
+			"repo": p.repo,
+			"item": file.Name,
+		})
+
+		if debug {
+			l.Debugln(p, "handling", file.Name)
+		}
+
+		switch {
+		case protocol.IsDirectory(file.Flags) && protocol.IsDeleted(file.Flags):
+			// A deleted directory
+			p.deleteDir(file)
+		case protocol.IsDirectory(file.Flags):
+			// A new or changed directory
+			p.handleDir(file)
+		case protocol.IsDeleted(file.Flags):
+			// A deleted file
+			p.deleteFile(file)
+		default:
+			// A new or changed file
+			p.handleFile(file, copyChan, pullChan)
+		}
+
+		changed++
+		return true
+	})
+
+	// Signal copy and puller routines that we are done with the in data for
+	// this iteration
+	close(copyChan)
+	close(pullChan)
+
+	// Wait for them to finish, then signal the finisher chan that there will
+	// be no more input.
+	wg.Wait()
+	close(finisherChan)
+
+	// Wait for the finisherChan to finish.
+	doneWg.Wait()
+
+	return changed
+}
+
+// handleDir creates or updates the given directory
+func (p *Puller) handleDir(file protocol.FileInfo) {
+	realName := filepath.Join(p.dir, file.Name)
+	mode := os.FileMode(file.Flags & 0777)
+
+	if debug {
+		curFile := p.model.CurrentRepoFile(p.repo, file.Name)
+		l.Debugf("need dir\n\t%v\n\t%v", file, curFile)
+	}
+
+	var err error
+	if info, err := os.Stat(realName); err != nil && os.IsNotExist(err) {
+		err = os.MkdirAll(realName, mode)
+	} else if !info.IsDir() {
+		l.Infof("Puller (repo %q, file %q): should be dir, but is not", p.repo, file.Name)
+		return
+	} else {
+		err = os.Chmod(realName, mode)
+	}
+
+	if err == nil {
+		p.model.updateLocal(p.repo, file)
+	}
+}
+
+// deleteDir attempts to delete the given directory
+func (p *Puller) deleteDir(file protocol.FileInfo) {
+	realName := filepath.Join(p.dir, file.Name)
+	err := os.Remove(realName)
+	if err == nil || os.IsNotExist(err) {
+		p.model.updateLocal(p.repo, file)
+	}
+}
+
+// deleteFile attempts to delete the given file
+func (p *Puller) deleteFile(file protocol.FileInfo) {
+	realName := filepath.Join(p.dir, file.Name)
+	realDir := filepath.Dir(realName)
+	if info, err := os.Stat(realDir); err == nil && info.IsDir() && info.Mode()&04 == 0 {
+		// A non-writeable directory (for this user; we assume that's the
+		// relevant part). Temporarily change the mode so we can delete the
+		// file inside it.
+		err = os.Chmod(realDir, 0755)
+		if err == nil {
+			defer func() {
+				err = os.Chmod(realDir, info.Mode())
+				if err != nil {
+					panic(err)
+				}
+			}()
+		}
+	}
+
+	var err error
+	if p.versioner != nil {
+		err = p.versioner.Archive(realName)
+	} else {
+		err = os.Remove(realName)
+	}
+
+	if err != nil {
+		l.Infof("Puller (repo %q, file %q): delete: %v", p.repo, file.Name, err)
+	} else {
+		p.model.updateLocal(p.repo, file)
+	}
+}
+
+// handleFile queues the copies and pulls as necessary for a single new or
+// changed file.
+func (p *Puller) handleFile(file protocol.FileInfo, copyChan chan<- copyBlocksState, pullChan chan<- pullBlockState) {
+	curFile := p.model.CurrentRepoFile(p.repo, file.Name)
+	copyBlocks, pullBlocks := scanner.BlockDiff(curFile.Blocks, file.Blocks)
+
+	if len(copyBlocks) == len(curFile.Blocks) && len(pullBlocks) == 0 {
+		// We are supposed to copy the entire file, and then fetch nothing. We
+		// are only updating metadata, so we don't actually *need* to make the
+		// copy.
+		if debug {
+			l.Debugln(p, "taking shortcut on", file.Name)
+		}
+		p.shortcutFile(file)
+		return
+	}
+
+	// Figure out the absolute filenames we need once and for all
+	tempName := filepath.Join(p.dir, defTempNamer.TempName(file.Name))
+	realName := filepath.Join(p.dir, file.Name)
+
+	s := sharedPullerState{
+		file:       file,
+		repo:       p.repo,
+		tempName:   tempName,
+		realName:   realName,
+		pullNeeded: len(pullBlocks),
+	}
+	if len(copyBlocks) > 0 {
+		s.copyNeeded = 1
+	}
+
+	if debug {
+		l.Debugf("%v need file %s; copy %d, pull %d", p, file.Name, len(copyBlocks), len(pullBlocks))
+	}
+
+	if len(copyBlocks) > 0 {
+		cs := copyBlocksState{
+			sharedPullerState: &s,
+			blocks:            copyBlocks,
+		}
+		copyChan <- cs
+	}
+
+	if len(pullBlocks) > 0 {
+		for _, block := range pullBlocks {
+			ps := pullBlockState{
+				sharedPullerState: &s,
+				block:             block,
+			}
+			pullChan <- ps
+		}
+	}
+}
+
+// shortcutFile sets file mode and modification time, when that's the only
+// thing that has changed.
+func (p *Puller) shortcutFile(file protocol.FileInfo) {
+	realName := filepath.Join(p.dir, file.Name)
+	err := os.Chmod(realName, os.FileMode(file.Flags&0777))
+	if err != nil {
+		l.Infof("Puller (repo %q, file %q): shortcut: %v", p.repo, file.Name, err)
+		return
+	}
+
+	t := time.Unix(file.Modified, 0)
+	err = os.Chtimes(realName, t, t)
+	if err != nil {
+		l.Infof("Puller (repo %q, file %q): shortcut: %v", p.repo, file.Name, err)
+		return
+	}
+
+	p.model.updateLocal(p.repo, file)
+}
+
+// copierRoutine reads pullerStates until the in channel closes and performs
+// the relevant copy.
+func (p *Puller) copierRoutine(in <-chan copyBlocksState, out chan<- *sharedPullerState) {
+	buf := make([]byte, scanner.StandardBlockSize)
+
+nextFile:
+	for state := range in {
+		dstFd, err := state.tempFile()
+		if err != nil {
+			// Nothing more to do for this failed file (the error was logged
+			// when it happened)
+			continue nextFile
+		}
+
+		srcFd, err := state.sourceFile()
+		if err != nil {
+			// As above
+			continue nextFile
+		}
+
+		for _, block := range state.blocks {
+			buf = buf[:int(block.Size)]
+
+			_, err = srcFd.ReadAt(buf, block.Offset)
+			if err != nil {
+				state.earlyClose("src read", err)
+				srcFd.Close()
+				continue nextFile
+			}
+
+			_, err = dstFd.WriteAt(buf, block.Offset)
+			if err != nil {
+				state.earlyClose("dst write", err)
+				srcFd.Close()
+				continue nextFile
+			}
+		}
+
+		srcFd.Close()
+		state.copyDone()
+		out <- state.sharedPullerState
+	}
+}
+
+func (p *Puller) pullerRoutine(in <-chan pullBlockState, out chan<- *sharedPullerState) {
+nextBlock:
+	for state := range in {
+		if state.failed() != nil {
+			continue nextBlock
+		}
+
+		// Select the least busy node to pull the block frop.model. If we found no
+		// feasible node at all, fail the block (and in the long run, the
+		// file).
+		potentialNodes := p.model.availability(p.repo, state.file.Name)
+		selected := activity.leastBusy(potentialNodes)
+		if selected == (protocol.NodeID{}) {
+			state.earlyClose("pull", errNoNode)
+			continue nextBlock
+		}
+
+		// Get an fd to the temporary file. Tehcnically we don't need it until
+		// after fetching the block, but if we run into an error here there is
+		// no point in issuing the request to the network.
+		fd, err := state.tempFile()
+		if err != nil {
+			continue nextBlock
+		}
+
+		// Fetch the block, while marking the selected node as in use so that
+		// leastBusy can select another node when someone else asks.
+		activity.using(selected)
+		buf, err := p.model.requestGlobal(selected, p.repo, state.file.Name, state.block.Offset, int(state.block.Size), state.block.Hash)
+		activity.done(selected)
+		if err != nil {
+			state.earlyClose("pull", err)
+			continue nextBlock
+		}
+
+		// Save the block data we got from the cluster
+		_, err = fd.WriteAt(buf, state.block.Offset)
+		if err != nil {
+			state.earlyClose("save", err)
+			continue nextBlock
+		}
+
+		state.pullDone()
+		out <- state.sharedPullerState
+	}
+}
+
+func (p *Puller) finisherRoutine(in <-chan *sharedPullerState) {
+	for state := range in {
+		if closed, err := state.finalClose(); closed {
+			if debug {
+				l.Debugln(p, "closing", state.file.Name)
+			}
+			if err != nil {
+				l.Warnln("puller: final:", err)
+				continue
+			}
+
+			// Verify the file against expected hashes
+			fd, err := os.Open(state.tempName)
+			if err != nil {
+				l.Warnln("puller: final:", err)
+				continue
+			}
+			err = scanner.Verify(fd, scanner.StandardBlockSize, state.file.Blocks)
+			fd.Close()
+			if err != nil {
+				l.Warnln("puller: final:", state.file.Name, err)
+				continue
+			}
+
+			// Set the correct permission bits on the new file
+			err = os.Chmod(state.tempName, os.FileMode(state.file.Flags&0777))
+			if err != nil {
+				os.Remove(state.tempName)
+				l.Warnln("puller: final:", err)
+				continue
+			}
+
+			// Set the correct timestamp on the new file
+			t := time.Unix(state.file.Modified, 0)
+			err = os.Chtimes(state.tempName, t, t)
+			if err != nil {
+				os.Remove(state.tempName)
+				l.Warnln("puller: final:", err)
+				continue
+			}
+
+			// 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 {
+				err = p.versioner.Archive(state.realName)
+				if err != nil {
+					os.Remove(state.tempName)
+					l.Warnln("puller: final:", err)
+					continue
+				}
+			}
+
+			// Replace the original file with the new one
+			err = osutil.Rename(state.tempName, state.realName)
+			if err != nil {
+				os.Remove(state.tempName)
+				l.Warnln("puller: final:", err)
+				continue
+			}
+
+			// Record the updated file in the index
+			p.model.updateLocal(p.repo, state.file)
+		}
+	}
+}
+
+// clean deletes orphaned temporary files
+func (p *Puller) clean() {
+	filepath.Walk(p.dir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if info.Mode().IsRegular() && defTempNamer.IsTemporary(path) {
+			os.Remove(path)
+		}
+
+		return nil
+	})
+}
+
+func invalidateRepo(cfg *config.Configuration, repoID string, err error) {
+	for i := range cfg.Repositories {
+		repo := &cfg.Repositories[i]
+		if repo.ID == repoID {
+			repo.Invalid = err.Error()
+			return
+		}
+	}
+}

internal/model/sharedpullerstate.go

@@ -0,0 +1,183 @@
+// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
+// All rights reserved. Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file.
+
+package model
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/syncthing/syncthing/internal/protocol"
+)
+
+// A sharedPullerState is kept for each file that is being synced and is kept
+// updated along the way.
+type sharedPullerState struct {
+	// Immutable, does not require locking
+	file     protocol.FileInfo
+	repo     string
+	tempName string
+	realName string
+
+	// Mutable, must be locked for access
+	err        error      // The first error we hit
+	fd         *os.File   // The fd of the temp file
+	copyNeeded int        // Number of copy actions we expect to happen
+	pullNeeded int        // Number of block pulls we expect to happen
+	closed     bool       // Set when the file has been closed
+	mut        sync.Mutex // Protects the above
+}
+
+// tempFile returns the fd for the temporary file, reusing an open fd
+// or creating the file as necessary.
+func (s *sharedPullerState) tempFile() (*os.File, error) {
+	s.mut.Lock()
+	defer s.mut.Unlock()
+
+	// If we've already hit an error, return early
+	if s.err != nil {
+		return nil, s.err
+	}
+
+	// If the temp file is already open, return the file descriptor
+	if s.fd != nil {
+		return s.fd, nil
+	}
+
+	// Ensure that the parent directory exists or can be created
+	dir := filepath.Dir(s.tempName)
+	if info, err := os.Stat(dir); err != nil && os.IsNotExist(err) {
+		err = os.MkdirAll(dir, 0755)
+		if err != nil {
+			s.earlyCloseLocked("dst mkdir", err)
+			return nil, err
+		}
+	} else if err != nil {
+		s.earlyCloseLocked("dst stat dir", err)
+		return nil, err
+	} else if !info.IsDir() {
+		err = fmt.Errorf("%q: not a directory", dir)
+		s.earlyCloseLocked("dst mkdir", err)
+		return nil, err
+	} else if info.Mode()&04 == 0 {
+		err := os.Chmod(dir, 0755)
+		if err == nil {
+			defer func() {
+				err := os.Chmod(dir, info.Mode().Perm())
+				if err != nil {
+					panic(err)
+				}
+			}()
+		}
+	}
+
+	// Attempt to create the temp file
+	fd, err := os.OpenFile(s.tempName, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0644)
+	if err != nil {
+		s.earlyCloseLocked("dst create", err)
+		return nil, err
+	}
+
+	// Same fd will be used by all writers
+	s.fd = fd
+
+	return fd, nil
+}
+
+// sourceFile opens the existing source file for reading
+func (s *sharedPullerState) sourceFile() (*os.File, error) {
+	s.mut.Lock()
+	defer s.mut.Unlock()
+
+	// If we've already hit an error, return early
+	if s.err != nil {
+		return nil, s.err
+	}
+
+	// Attempt to open the existing file
+	fd, err := os.Open(s.realName)
+	if err != nil {
+		s.earlyCloseLocked("src open", err)
+		return nil, err
+	}
+
+	return fd, nil
+}
+
+// earlyClose prints a warning message composed of the context and
+// error, and marks the sharedPullerState as failed. Is a no-op when called on
+// an already failed state.
+func (s *sharedPullerState) earlyClose(context string, err error) {
+	s.mut.Lock()
+	defer s.mut.Unlock()
+
+	s.earlyCloseLocked(context, err)
+}
+
+func (s *sharedPullerState) earlyCloseLocked(context string, err error) {
+	if s.err != nil {
+		return
+	}
+
+	l.Infof("Puller (repo %q, file %q): %s: %v", s.repo, s.file.Name, context, err)
+	s.err = err
+	if s.fd != nil {
+		s.fd.Close()
+		os.Remove(s.tempName)
+	}
+	s.closed = true
+}
+
+func (s *sharedPullerState) failed() error {
+	s.mut.Lock()
+	defer s.mut.Unlock()
+
+	return s.err
+}
+
+func (s *sharedPullerState) copyDone() {
+	s.mut.Lock()
+	s.copyNeeded--
+	if debug {
+		l.Debugln("sharedPullerState", s.repo, s.file.Name, "copyNeeded ->", s.pullNeeded)
+	}
+	s.mut.Unlock()
+}
+
+func (s *sharedPullerState) pullDone() {
+	s.mut.Lock()
+	s.pullNeeded--
+	if debug {
+		l.Debugln("sharedPullerState", s.repo, s.file.Name, "pullNeeded ->", s.pullNeeded)
+	}
+	s.mut.Unlock()
+}
+
+// finalClose atomically closes and returns closed status of a file. A true
+// first return value means the file was closed and should be finished, with
+// the error indicating the success or failure of the close. A false first
+// return value indicates the file is not ready to be closed, or is already
+// closed and should in either case not be finished off now.
+func (s *sharedPullerState) finalClose() (bool, error) {
+	s.mut.Lock()
+	defer s.mut.Unlock()
+
+	if s.pullNeeded+s.copyNeeded != 0 {
+		// Not done yet.
+		return false, nil
+	}
+	if s.closed {
+		// Already handled.
+		return false, nil
+	}
+
+	s.closed = true
+	if fd := s.fd; fd != nil {
+		s.fd = nil
+		return true, fd.Close()
+	}
+	return true, nil
+}

internal/model/sharedpullerstate_test.go

@@ -0,0 +1,52 @@
+// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
+// All rights reserved. Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file.
+
+package model
+
+import "testing"
+
+func TestSourceFileOK(t *testing.T) {
+	s := sharedPullerState{
+		realName: "testdata/foo",
+	}
+
+	fd, err := s.sourceFile()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fd == nil {
+		t.Fatal("Unexpected nil fd")
+	}
+
+	bs := make([]byte, 6)
+	n, err := fd.Read(bs)
+
+	if n != len(bs) {
+		t.Fatal("Wrong read length %d != %d", n, len(bs))
+	}
+	if string(bs) != "foobar" {
+		t.Fatal("Wrong contents %s != foobar", bs)
+	}
+
+	if err := s.failed(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSourceFileBad(t *testing.T) {
+	s := sharedPullerState{
+		realName: "nonexistent",
+	}
+
+	fd, err := s.sourceFile()
+	if err == nil {
+		t.Fatal("Unexpected nil error")
+	}
+	if fd != nil {
+		t.Fatal("Unexpected non-nil fd")
+	}
+	if err := s.failed(); err == nil {
+		t.Fatal("Unexpected nil failed()")
+	}
+}

internal/protocol/debug.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/syncthing/syncthing/logger"
+	"github.com/syncthing/syncthing/internal/logger"
 )
 
 var (

internal/protocol/nodeid.go

@@ -13,7 +13,7 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/syncthing/syncthing/luhn"
+	"github.com/syncthing/syncthing/internal/luhn"
 )
 
 type NodeID [32]byte