Compare commits
17 Commits
v1.4.0-rc.3
...
v1.4.0
| Author | SHA1 | Date | |
|---|---|---|---|
| db02545ef3 | |||
| 52e72e0122 | |||
| d1e0a38c04 | |||
| 0b610017ea | |||
| 9a1df97c69 | |||
| ee61da5b6a | |||
| a5e12a0a3d | |||
| 4f29180e7c | |||
| 0fb2cd52ff | |||
| a4bd4d118a | |||
| bb375b1aff | |||
| 05e23f1991 | |||
| 71de6fe290 | |||
| 6a840a040b | |||
| c3637f2191 | |||
| ca90f4e6af | |||
| 51fa36d61f |
@@ -654,7 +654,11 @@ func shouldBuildSyso(dir string) (string, error) {
|
||||
}
|
||||
|
||||
jsonPath := filepath.Join(dir, "versioninfo.json")
|
||||
ioutil.WriteFile(jsonPath, bs, 0644)
|
||||
err = ioutil.WriteFile(jsonPath, bs, 0644)
|
||||
if err != nil {
|
||||
return "", errors.New("failed to create " + jsonPath + ": " + err.Error())
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := os.Remove(jsonPath); err != nil {
|
||||
log.Printf("Warning: unable to remove generated %s: %v. Please remove it manually.", jsonPath, err)
|
||||
@@ -860,13 +864,22 @@ func getVersion() string {
|
||||
return "unknown-dev"
|
||||
}
|
||||
|
||||
func semanticVersion() (major, minor, patch, build string) {
|
||||
func semanticVersion() (major, minor, patch, build int) {
|
||||
r := regexp.MustCompile(`v(?P<Major>\d+)\.(?P<Minor>\d+).(?P<Patch>\d+).*\+(?P<CommitsAhead>\d+)`)
|
||||
matches := r.FindStringSubmatch(getVersion())
|
||||
if len(matches) != 5 {
|
||||
return "0", "0", "0", "0"
|
||||
return 0, 0, 0, 0
|
||||
}
|
||||
return matches[1], matches[2], matches[3], matches[4]
|
||||
|
||||
var ints [4]int
|
||||
for i := 1; i < 5; i++ {
|
||||
value, err := strconv.Atoi(matches[i])
|
||||
if err != nil {
|
||||
return 0, 0, 0, 0
|
||||
}
|
||||
ints[i-1] = value
|
||||
}
|
||||
return ints[0], ints[1], ints[2], ints[3]
|
||||
}
|
||||
|
||||
func getBranchSuffix() string {
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"flag"
|
||||
@@ -95,7 +96,7 @@ func checkServer(deviceID protocol.DeviceID, server string) checkResult {
|
||||
})
|
||||
|
||||
go func() {
|
||||
addresses, err := disco.Lookup(deviceID)
|
||||
addresses, err := disco.Lookup(context.Background(), deviceID)
|
||||
res <- checkResult{addresses: addresses, error: err}
|
||||
}()
|
||||
|
||||
|
||||
@@ -148,7 +148,7 @@ func idxck(ldb backend.Backend) (success bool) {
|
||||
}
|
||||
}
|
||||
|
||||
if fi.BlocksHash != nil {
|
||||
if len(fi.Blocks) == 0 && len(fi.BlocksHash) != 0 {
|
||||
key := string(fi.BlocksHash)
|
||||
if _, ok := blocklists[key]; !ok {
|
||||
fmt.Printf("Missing block list for file %q, block list hash %x\n", fi.Name, fi.BlocksHash)
|
||||
|
||||
@@ -119,8 +119,8 @@ are mostly useful for developers. Use with care.
|
||||
"h", "m" and "s" abbreviations for hours minutes and seconds.
|
||||
Valid values are like "720h", "30s", etc.
|
||||
|
||||
STGCBLOCKSEVERY Set to a time interval to override the default database
|
||||
block GC interval of 13 hours. Same format as the
|
||||
STGCINDIRECTEVERY Set to a time interval to override the default database
|
||||
indirection GC interval of 13 hours. Same format as the
|
||||
STRECHECKDBEVERY variable.
|
||||
|
||||
GOMAXPROCS Set the maximum number of CPU cores to use. Defaults to all
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/discover"
|
||||
@@ -26,7 +27,7 @@ func (m *mockedCachingMux) Stop() {
|
||||
|
||||
// from events.Finder
|
||||
|
||||
func (m *mockedCachingMux) Lookup(deviceID protocol.DeviceID) (direct []string, err error) {
|
||||
func (m *mockedCachingMux) Lookup(ctx context.Context, deviceID protocol.DeviceID) (direct []string, err error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -360,6 +360,12 @@ func (s *service) connect(ctx context.Context) {
|
||||
var seen []string
|
||||
|
||||
for _, deviceCfg := range cfg.Devices {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
deviceID := deviceCfg.DeviceID
|
||||
if deviceID == s.myID {
|
||||
continue
|
||||
@@ -380,7 +386,7 @@ func (s *service) connect(ctx context.Context) {
|
||||
for _, addr := range deviceCfg.Addresses {
|
||||
if addr == "dynamic" {
|
||||
if s.discoverer != nil {
|
||||
if t, err := s.discoverer.Lookup(deviceID); err == nil {
|
||||
if t, err := s.discoverer.Lookup(ctx, deviceID); err == nil {
|
||||
addrs = append(addrs, t...)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,10 +48,15 @@ type ReadTransaction interface {
|
||||
// purposes of saving memory when transactions are in-RAM. Note that
|
||||
// transactions may be checkpointed *anyway* even if this is not called, due to
|
||||
// resource constraints, but this gives you a chance to decide when.
|
||||
//
|
||||
// Functions can be passed to Checkpoint. These are run if and only if the
|
||||
// checkpoint will result in a flush, and will run before the flush. The
|
||||
// transaction can be accessed via a closure. If an error is returned from
|
||||
// these functions the flush will be aborted and the error bubbled.
|
||||
type WriteTransaction interface {
|
||||
ReadTransaction
|
||||
Writer
|
||||
Checkpoint() error
|
||||
Checkpoint(...func() error) error
|
||||
Commit() error
|
||||
}
|
||||
|
||||
|
||||
@@ -150,8 +150,8 @@ func (t *leveldbTransaction) Put(key, val []byte) error {
|
||||
return t.checkFlush(dbFlushBatchMax)
|
||||
}
|
||||
|
||||
func (t *leveldbTransaction) Checkpoint() error {
|
||||
return t.checkFlush(dbFlushBatchMin)
|
||||
func (t *leveldbTransaction) Checkpoint(preFlush ...func() error) error {
|
||||
return t.checkFlush(dbFlushBatchMin, preFlush...)
|
||||
}
|
||||
|
||||
func (t *leveldbTransaction) Commit() error {
|
||||
@@ -167,10 +167,15 @@ func (t *leveldbTransaction) Release() {
|
||||
}
|
||||
|
||||
// checkFlush flushes and resets the batch if its size exceeds the given size.
|
||||
func (t *leveldbTransaction) checkFlush(size int) error {
|
||||
func (t *leveldbTransaction) checkFlush(size int, preFlush ...func() error) error {
|
||||
if len(t.batch.Dump()) < size {
|
||||
return nil
|
||||
}
|
||||
for _, hook := range preFlush {
|
||||
if err := hook(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return t.flush()
|
||||
}
|
||||
|
||||
|
||||
+70
-41
@@ -19,22 +19,25 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// We set the bloom filter capacity to handle 100k individual block lists
|
||||
// with a false positive probability of 1% for the first pass. Once we know
|
||||
// how many block lists we have we will use that number instead, if it's
|
||||
// more than 100k. For fewer than 100k block lists we will just get better
|
||||
// false positive rate instead.
|
||||
blockGCBloomCapacity = 100000
|
||||
blockGCBloomFalsePositiveRate = 0.01 // 1%
|
||||
blockGCDefaultInterval = 13 * time.Hour
|
||||
blockGCTimeKey = "lastBlockGCTime"
|
||||
// We set the bloom filter capacity to handle 100k individual items with
|
||||
// a false positive probability of 1% for the first pass. Once we know
|
||||
// how many items we have we will use that number instead, if it's more
|
||||
// than 100k. For fewer than 100k items we will just get better false
|
||||
// positive rate instead.
|
||||
indirectGCBloomCapacity = 100000
|
||||
indirectGCBloomFalsePositiveRate = 0.01 // 1%
|
||||
indirectGCDefaultInterval = 13 * time.Hour
|
||||
indirectGCTimeKey = "lastIndirectGCTime"
|
||||
|
||||
// Use indirection for the block list when it exceeds this many entries
|
||||
blocksIndirectionCutoff = 3
|
||||
)
|
||||
|
||||
var blockGCInterval = blockGCDefaultInterval
|
||||
var indirectGCInterval = indirectGCDefaultInterval
|
||||
|
||||
func init() {
|
||||
if dur, err := time.ParseDuration(os.Getenv("STGCBLOCKSEVERY")); err == nil {
|
||||
blockGCInterval = dur
|
||||
if dur, err := time.ParseDuration(os.Getenv("STGCINDIRECTEVERY")); err == nil {
|
||||
indirectGCInterval = dur
|
||||
}
|
||||
}
|
||||
|
||||
@@ -124,11 +127,17 @@ func (db *Lowlevel) updateRemoteFiles(folder, device []byte, fs []protocol.FileI
|
||||
return err
|
||||
}
|
||||
|
||||
if err := t.Checkpoint(); err != nil {
|
||||
if err := t.Checkpoint(func() error {
|
||||
return meta.toDB(t, folder)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err := meta.toDB(t, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return t.Commit()
|
||||
}
|
||||
|
||||
@@ -227,11 +236,17 @@ func (db *Lowlevel) updateLocalFiles(folder []byte, fs []protocol.FileInfo, meta
|
||||
}
|
||||
}
|
||||
|
||||
if err := t.Checkpoint(); err != nil {
|
||||
if err := t.Checkpoint(func() error {
|
||||
return meta.toDB(t, folder)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err := meta.toDB(t, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return t.Commit()
|
||||
}
|
||||
|
||||
@@ -473,18 +488,28 @@ func (db *Lowlevel) dropPrefix(prefix []byte) error {
|
||||
}
|
||||
|
||||
func (db *Lowlevel) gcRunner() {
|
||||
t := time.NewTimer(db.timeUntil(blockGCTimeKey, blockGCInterval))
|
||||
// Calculate the time for the next GC run. Even if we should run GC
|
||||
// directly, give the system a while to get up and running and do other
|
||||
// stuff first. (We might have migrations and stuff which would be
|
||||
// better off running before GC.)
|
||||
next := db.timeUntil(indirectGCTimeKey, indirectGCInterval)
|
||||
if next < time.Minute {
|
||||
next = time.Minute
|
||||
}
|
||||
|
||||
t := time.NewTimer(next)
|
||||
defer t.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-db.gcStop:
|
||||
return
|
||||
case <-t.C:
|
||||
if err := db.gcBlocks(); err != nil {
|
||||
l.Warnln("Database block GC failed:", err)
|
||||
if err := db.gcIndirect(); err != nil {
|
||||
l.Warnln("Database indirection GC failed:", err)
|
||||
}
|
||||
db.recordTime(blockGCTimeKey)
|
||||
t.Reset(db.timeUntil(blockGCTimeKey, blockGCInterval))
|
||||
db.recordTime(indirectGCTimeKey)
|
||||
t.Reset(db.timeUntil(indirectGCTimeKey, indirectGCInterval))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -509,15 +534,16 @@ func (db *Lowlevel) timeUntil(key string, every time.Duration) time.Duration {
|
||||
return sleepTime
|
||||
}
|
||||
|
||||
func (db *Lowlevel) gcBlocks() error {
|
||||
// The block GC uses a bloom filter to track used block lists. This means
|
||||
// iterating over all items, adding their block lists to the filter, then
|
||||
// iterating over the block lists and removing those that don't match the
|
||||
// filter. The filter will give false positives so we will keep around one
|
||||
// percent of block lists that we don't really need (at most).
|
||||
func (db *Lowlevel) gcIndirect() error {
|
||||
// The indirection GC uses bloom filters to track used block lists and
|
||||
// versions. This means iterating over all items, adding their hashes to
|
||||
// the filter, then iterating over the indirected items and removing
|
||||
// those that don't match the filter. The filter will give false
|
||||
// positives so we will keep around one percent of things that we don't
|
||||
// really need (at most).
|
||||
//
|
||||
// Block GC needs to run when there are no modifications to the FileInfos or
|
||||
// block lists.
|
||||
// Indirection GC needs to run when there are no modifications to the
|
||||
// FileInfos or indirected items.
|
||||
|
||||
db.gcMut.Lock()
|
||||
defer db.gcMut.Unlock()
|
||||
@@ -528,30 +554,32 @@ func (db *Lowlevel) gcBlocks() error {
|
||||
}
|
||||
defer t.Release()
|
||||
|
||||
// Set up the bloom filter with the initial capacity and false positive
|
||||
// rate, or higher capacity if we've done this before and seen lots of block
|
||||
// lists.
|
||||
// Set up the bloom filters with the initial capacity and false positive
|
||||
// rate, or higher capacity if we've done this before and seen lots of
|
||||
// items. For simplicity's sake we track just one count, which is the
|
||||
// highest of the various indirected items.
|
||||
|
||||
capacity := blockGCBloomCapacity
|
||||
capacity := indirectGCBloomCapacity
|
||||
if db.gcKeyCount > capacity {
|
||||
capacity = db.gcKeyCount
|
||||
}
|
||||
filter := bloom.NewWithEstimates(uint(capacity), blockGCBloomFalsePositiveRate)
|
||||
blockFilter := bloom.NewWithEstimates(uint(capacity), indirectGCBloomFalsePositiveRate)
|
||||
|
||||
// Iterate the FileInfos, unmarshal the blocks hashes and add them to
|
||||
// the filter.
|
||||
// Iterate the FileInfos, unmarshal the block and version hashes and
|
||||
// add them to the filter.
|
||||
|
||||
it, err := db.NewPrefixIterator([]byte{KeyTypeDevice})
|
||||
it, err := t.NewPrefixIterator([]byte{KeyTypeDevice})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer it.Release()
|
||||
for it.Next() {
|
||||
var bl BlocksHashOnly
|
||||
if err := bl.Unmarshal(it.Value()); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(bl.BlocksHash) > 0 {
|
||||
filter.Add(bl.BlocksHash)
|
||||
blockFilter.Add(bl.BlocksHash)
|
||||
}
|
||||
}
|
||||
it.Release()
|
||||
@@ -562,15 +590,16 @@ func (db *Lowlevel) gcBlocks() error {
|
||||
// Iterate over block lists, removing keys with hashes that don't match
|
||||
// the filter.
|
||||
|
||||
it, err = db.NewPrefixIterator([]byte{KeyTypeBlockList})
|
||||
it, err = t.NewPrefixIterator([]byte{KeyTypeBlockList})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
matched := 0
|
||||
defer it.Release()
|
||||
matchedBlocks := 0
|
||||
for it.Next() {
|
||||
key := blockListKey(it.Key())
|
||||
if filter.Test(key.BlocksHash()) {
|
||||
matched++
|
||||
if blockFilter.Test(key.BlocksHash()) {
|
||||
matchedBlocks++
|
||||
continue
|
||||
}
|
||||
if err := t.Delete(key); err != nil {
|
||||
@@ -583,7 +612,7 @@ func (db *Lowlevel) gcBlocks() error {
|
||||
}
|
||||
|
||||
// Remember the number of unique keys we kept until the next pass.
|
||||
db.gcKeyCount = matched
|
||||
db.gcKeyCount = matchedBlocks
|
||||
|
||||
if err := t.Commit(); err != nil {
|
||||
return err
|
||||
|
||||
+16
-12
@@ -28,8 +28,8 @@ type metadataTracker struct {
|
||||
}
|
||||
|
||||
type metaKey struct {
|
||||
dev protocol.DeviceID
|
||||
flags uint32
|
||||
dev protocol.DeviceID
|
||||
flag uint32
|
||||
}
|
||||
|
||||
func newMetadataTracker() *metadataTracker {
|
||||
@@ -62,8 +62,8 @@ func (m *metadataTracker) Marshal() ([]byte, error) {
|
||||
|
||||
// toDB saves the marshalled metadataTracker to the given db, under the key
|
||||
// corresponding to the given folder
|
||||
func (m *metadataTracker) toDB(db *Lowlevel, folder []byte) error {
|
||||
key, err := db.keyer.GenerateFolderMetaKey(nil, folder)
|
||||
func (m *metadataTracker) toDB(t readWriteTransaction, folder []byte) error {
|
||||
key, err := t.keyer.GenerateFolderMetaKey(nil, folder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -79,7 +79,7 @@ func (m *metadataTracker) toDB(db *Lowlevel, folder []byte) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = db.Put(key, bs)
|
||||
err = t.Put(key, bs)
|
||||
if err == nil {
|
||||
m.dirty = false
|
||||
}
|
||||
@@ -103,14 +103,18 @@ func (m *metadataTracker) fromDB(db *Lowlevel, folder []byte) error {
|
||||
|
||||
// countsPtr returns a pointer to the corresponding Counts struct, if
|
||||
// necessary allocating one in the process
|
||||
func (m *metadataTracker) countsPtr(dev protocol.DeviceID, flags uint32) *Counts {
|
||||
func (m *metadataTracker) countsPtr(dev protocol.DeviceID, flag uint32) *Counts {
|
||||
// must be called with the mutex held
|
||||
|
||||
key := metaKey{dev, flags}
|
||||
if bits.OnesCount32(flag) > 1 {
|
||||
panic("incorrect usage: set at most one bit in flag")
|
||||
}
|
||||
|
||||
key := metaKey{dev, flag}
|
||||
idx, ok := m.indexes[key]
|
||||
if !ok {
|
||||
idx = len(m.counts.Counts)
|
||||
m.counts.Counts = append(m.counts.Counts, Counts{DeviceID: dev[:], LocalFlags: flags})
|
||||
m.counts.Counts = append(m.counts.Counts, Counts{DeviceID: dev[:], LocalFlags: flag})
|
||||
m.indexes[key] = idx
|
||||
}
|
||||
return &m.counts.Counts[idx]
|
||||
@@ -157,8 +161,8 @@ func (m *metadataTracker) updateSeqLocked(dev protocol.DeviceID, f FileIntf) {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *metadataTracker) addFileLocked(dev protocol.DeviceID, flags uint32, f FileIntf) {
|
||||
cp := m.countsPtr(dev, flags)
|
||||
func (m *metadataTracker) addFileLocked(dev protocol.DeviceID, flag uint32, f FileIntf) {
|
||||
cp := m.countsPtr(dev, flag)
|
||||
|
||||
switch {
|
||||
case f.IsDeleted():
|
||||
@@ -196,8 +200,8 @@ func (m *metadataTracker) removeFile(dev protocol.DeviceID, f FileIntf) {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *metadataTracker) removeFileLocked(dev protocol.DeviceID, flags uint32, f FileIntf) {
|
||||
cp := m.countsPtr(dev, f.FileLocalFlags())
|
||||
func (m *metadataTracker) removeFileLocked(dev protocol.DeviceID, flag uint32, f FileIntf) {
|
||||
cp := m.countsPtr(dev, flag)
|
||||
|
||||
switch {
|
||||
case f.IsDeleted():
|
||||
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
"sort"
|
||||
"testing"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/db/backend"
|
||||
"github.com/syncthing/syncthing/lib/fs"
|
||||
"github.com/syncthing/syncthing/lib/protocol"
|
||||
)
|
||||
|
||||
@@ -101,3 +103,75 @@ func TestMetaSequences(t *testing.T) {
|
||||
t.Error("sequence of first device should be 4, not", seq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecalcMeta(t *testing.T) {
|
||||
ldb := NewLowlevel(backend.OpenMemory())
|
||||
defer ldb.Close()
|
||||
|
||||
// Add some files
|
||||
s1 := NewFileSet("test", fs.NewFilesystem(fs.FilesystemTypeFake, "fake"), ldb)
|
||||
files := []protocol.FileInfo{
|
||||
{Name: "a", Size: 1000},
|
||||
{Name: "b", Size: 2000},
|
||||
}
|
||||
s1.Update(protocol.LocalDeviceID, files)
|
||||
|
||||
// Verify local/global size
|
||||
snap := s1.Snapshot()
|
||||
ls := snap.LocalSize()
|
||||
gs := snap.GlobalSize()
|
||||
snap.Release()
|
||||
if ls.Bytes != 3000 {
|
||||
t.Fatalf("Wrong initial local byte count, %d != 3000", ls.Bytes)
|
||||
}
|
||||
if gs.Bytes != 3000 {
|
||||
t.Fatalf("Wrong initial global byte count, %d != 3000", gs.Bytes)
|
||||
}
|
||||
|
||||
// Reach into the database to make the metadata tracker intentionally
|
||||
// wrong and out of date
|
||||
curSeq := s1.meta.Sequence(protocol.LocalDeviceID)
|
||||
tran, err := ldb.newReadWriteTransaction()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s1.meta.mut.Lock()
|
||||
s1.meta.countsPtr(protocol.LocalDeviceID, 0).Sequence = curSeq - 1 // too low
|
||||
s1.meta.countsPtr(protocol.LocalDeviceID, 0).Bytes = 1234 // wrong
|
||||
s1.meta.countsPtr(protocol.GlobalDeviceID, 0).Bytes = 1234 // wrong
|
||||
s1.meta.dirty = true
|
||||
s1.meta.mut.Unlock()
|
||||
if err := s1.meta.toDB(tran, []byte("test")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tran.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify that our bad data "took"
|
||||
snap = s1.Snapshot()
|
||||
ls = snap.LocalSize()
|
||||
gs = snap.GlobalSize()
|
||||
snap.Release()
|
||||
if ls.Bytes != 1234 {
|
||||
t.Fatalf("Wrong changed local byte count, %d != 1234", ls.Bytes)
|
||||
}
|
||||
if gs.Bytes != 1234 {
|
||||
t.Fatalf("Wrong changed global byte count, %d != 1234", gs.Bytes)
|
||||
}
|
||||
|
||||
// Create a new fileset, which will realize the inconsistency and recalculate
|
||||
s2 := NewFileSet("test", fs.NewFilesystem(fs.FilesystemTypeFake, "fake"), ldb)
|
||||
|
||||
// Verify local/global size
|
||||
snap = s2.Snapshot()
|
||||
ls = snap.LocalSize()
|
||||
gs = snap.GlobalSize()
|
||||
snap.Release()
|
||||
if ls.Bytes != 3000 {
|
||||
t.Fatalf("Wrong fixed local byte count, %d != 3000", ls.Bytes)
|
||||
}
|
||||
if gs.Bytes != 3000 {
|
||||
t.Fatalf("Wrong fixed global byte count, %d != 3000", gs.Bytes)
|
||||
}
|
||||
}
|
||||
|
||||
+80
-6
@@ -7,9 +7,11 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/db/backend"
|
||||
"github.com/syncthing/syncthing/lib/protocol"
|
||||
)
|
||||
|
||||
@@ -23,11 +25,17 @@ import (
|
||||
// 6: v0.14.50
|
||||
// 7: v0.14.53
|
||||
// 8: v1.4.0
|
||||
// 9: v1.4.0
|
||||
const (
|
||||
dbVersion = 8
|
||||
dbVersion = 9
|
||||
dbMinSyncthingVersion = "v1.4.0"
|
||||
)
|
||||
|
||||
var (
|
||||
errFolderIdxMissing = fmt.Errorf("folder db index missing")
|
||||
errDeviceIdxMissing = fmt.Errorf("device db index missing")
|
||||
)
|
||||
|
||||
type databaseDowngradeError struct {
|
||||
minSyncthingVersion string
|
||||
}
|
||||
@@ -49,6 +57,11 @@ type schemaUpdater struct {
|
||||
}
|
||||
|
||||
func (db *schemaUpdater) updateSchema() error {
|
||||
// Updating the schema can touch any and all parts of the database. Make
|
||||
// sure we do not run GC concurrently with schema migrations.
|
||||
db.gcMut.Lock()
|
||||
defer db.gcMut.Unlock()
|
||||
|
||||
miscDB := NewMiscDataNamespace(db.Lowlevel)
|
||||
prevVersion, _, err := miscDB.Int64("dbVersion")
|
||||
if err != nil {
|
||||
@@ -80,7 +93,7 @@ func (db *schemaUpdater) updateSchema() error {
|
||||
{5, db.updateSchemaTo5},
|
||||
{6, db.updateSchema5to6},
|
||||
{7, db.updateSchema6to7},
|
||||
{8, db.updateSchema7to8},
|
||||
{9, db.updateSchemato9},
|
||||
}
|
||||
|
||||
for _, m := range migrations {
|
||||
@@ -421,8 +434,9 @@ func (db *schemaUpdater) updateSchema6to7(_ int) error {
|
||||
return t.Commit()
|
||||
}
|
||||
|
||||
func (db *schemaUpdater) updateSchema7to8(_ int) error {
|
||||
func (db *schemaUpdater) updateSchemato9(prev int) error {
|
||||
// Loads and rewrites all files with blocks, to deduplicate block lists.
|
||||
// Checks for missing or incorrect sequence entries and rewrites those.
|
||||
|
||||
t, err := db.newReadWriteTransaction()
|
||||
if err != nil {
|
||||
@@ -430,15 +444,69 @@ func (db *schemaUpdater) updateSchema7to8(_ int) error {
|
||||
}
|
||||
defer t.close()
|
||||
|
||||
var sk []byte
|
||||
it, err := t.NewPrefixIterator([]byte{KeyTypeDevice})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
metas := make(map[string]*metadataTracker)
|
||||
for it.Next() {
|
||||
var fi protocol.FileInfo
|
||||
if err := fi.Unmarshal(it.Value()); err != nil {
|
||||
intf, err := t.unmarshalTrunc(it.Value(), false)
|
||||
if backend.IsNotFound(err) {
|
||||
// Unmarshal error due to missing parts (block list), probably
|
||||
// due to a bad migration in a previous RC. Drop this key, as
|
||||
// getFile would anyway return this as a "not found" in the
|
||||
// normal flow of things.
|
||||
if err := t.Delete(it.Key()); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
fi := intf.(protocol.FileInfo)
|
||||
device, ok := t.keyer.DeviceFromDeviceFileKey(it.Key())
|
||||
if !ok {
|
||||
return errDeviceIdxMissing
|
||||
}
|
||||
if bytes.Equal(device, protocol.LocalDeviceID[:]) {
|
||||
folder, ok := t.keyer.FolderFromDeviceFileKey(it.Key())
|
||||
if !ok {
|
||||
return errFolderIdxMissing
|
||||
}
|
||||
if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
|
||||
return err
|
||||
}
|
||||
switch dk, err := t.Get(sk); {
|
||||
case err != nil:
|
||||
if !backend.IsNotFound(err) {
|
||||
return err
|
||||
}
|
||||
fallthrough
|
||||
case !bytes.Equal(it.Key(), dk):
|
||||
folderStr := string(folder)
|
||||
meta, ok := metas[folderStr]
|
||||
if !ok {
|
||||
meta = loadMetadataTracker(db.Lowlevel, folderStr)
|
||||
metas[folderStr] = meta
|
||||
}
|
||||
fi.Sequence = meta.nextLocalSeq()
|
||||
if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := t.Put(sk, it.Key()); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := t.putFile(it.Key(), fi); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
if prev == 8 {
|
||||
// The transition to 8 already did the changes below.
|
||||
continue
|
||||
}
|
||||
if fi.Blocks == nil {
|
||||
continue
|
||||
}
|
||||
@@ -451,7 +519,13 @@ func (db *schemaUpdater) updateSchema7to8(_ int) error {
|
||||
return err
|
||||
}
|
||||
|
||||
db.recordTime(blockGCTimeKey)
|
||||
for folder, meta := range metas {
|
||||
if err := meta.toDB(t, []byte(folder)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
db.recordTime(indirectGCTimeKey)
|
||||
|
||||
return t.Commit()
|
||||
}
|
||||
|
||||
+95
-39
@@ -71,56 +71,105 @@ func init() {
|
||||
}
|
||||
|
||||
func NewFileSet(folder string, fs fs.Filesystem, db *Lowlevel) *FileSet {
|
||||
var s = FileSet{
|
||||
return &FileSet{
|
||||
folder: folder,
|
||||
fs: fs,
|
||||
db: db,
|
||||
meta: newMetadataTracker(),
|
||||
meta: loadMetadataTracker(db, folder),
|
||||
updateMutex: sync.NewMutex(),
|
||||
}
|
||||
|
||||
if err := s.meta.fromDB(db, []byte(folder)); err != nil {
|
||||
l.Infof("No stored folder metadata for %q: recalculating", folder)
|
||||
if err := s.recalcCounts(); backend.IsClosed(err) {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
} else if age := time.Since(s.meta.Created()); age > databaseRecheckInterval {
|
||||
l.Infof("Stored folder metadata for %q is %v old; recalculating", folder, age)
|
||||
if err := s.recalcCounts(); backend.IsClosed(err) {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
return &s
|
||||
}
|
||||
|
||||
func (s *FileSet) recalcCounts() error {
|
||||
s.meta = newMetadataTracker()
|
||||
|
||||
if err := s.db.checkGlobals([]byte(s.folder), s.meta); err != nil {
|
||||
return err
|
||||
func loadMetadataTracker(db *Lowlevel, folder string) *metadataTracker {
|
||||
recalc := func() *metadataTracker {
|
||||
meta, err := recalcMeta(db, folder)
|
||||
if backend.IsClosed(err) {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return meta
|
||||
}
|
||||
|
||||
t, err := s.db.newReadWriteTransaction()
|
||||
meta := newMetadataTracker()
|
||||
if err := meta.fromDB(db, []byte(folder)); err != nil {
|
||||
l.Infof("No stored folder metadata for %q; recalculating", folder)
|
||||
return recalc()
|
||||
}
|
||||
|
||||
curSeq := meta.Sequence(protocol.LocalDeviceID)
|
||||
if metaOK := verifyLocalSequence(curSeq, db, folder); !metaOK {
|
||||
l.Infof("Stored folder metadata for %q is out of date after crash; recalculating", folder)
|
||||
return recalc()
|
||||
}
|
||||
|
||||
if age := time.Since(meta.Created()); age > databaseRecheckInterval {
|
||||
l.Infof("Stored folder metadata for %q is %v old; recalculating", folder, age)
|
||||
return recalc()
|
||||
}
|
||||
|
||||
return meta
|
||||
}
|
||||
|
||||
func recalcMeta(db *Lowlevel, folder string) (*metadataTracker, error) {
|
||||
meta := newMetadataTracker()
|
||||
if err := db.checkGlobals([]byte(folder), meta); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
t, err := db.newReadWriteTransaction()
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
defer t.close()
|
||||
|
||||
var deviceID protocol.DeviceID
|
||||
err = t.withAllFolderTruncated([]byte(s.folder), func(device []byte, f FileInfoTruncated) bool {
|
||||
err = t.withAllFolderTruncated([]byte(folder), func(device []byte, f FileInfoTruncated) bool {
|
||||
copy(deviceID[:], device)
|
||||
s.meta.addFile(deviceID, f)
|
||||
meta.addFile(deviceID, f)
|
||||
return true
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.meta.SetCreated()
|
||||
return s.meta.toDB(s.db, []byte(s.folder))
|
||||
meta.SetCreated()
|
||||
if err := meta.toDB(t, []byte(folder)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := t.Commit(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// Verify the local sequence number from actual sequence entries. Returns
|
||||
// true if it was all good, or false if a fixup was necessary.
|
||||
func verifyLocalSequence(curSeq int64, db *Lowlevel, folder string) bool {
|
||||
// Walk the sequence index from the current (supposedly) highest
|
||||
// sequence number and raise the alarm if we get anything. This recovers
|
||||
// from the occasion where we have written sequence entries to disk but
|
||||
// not yet written new metadata to disk.
|
||||
//
|
||||
// Note that we can have the same thing happen for remote devices but
|
||||
// there it's not a problem -- we'll simply advertise a lower sequence
|
||||
// number than we've actually seen and receive some duplicate updates
|
||||
// and then be in sync again.
|
||||
|
||||
t, err := db.newReadOnlyTransaction()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
ok := true
|
||||
if err := t.withHaveSequence([]byte(folder), curSeq+1, func(fi FileIntf) bool {
|
||||
ok = false // we got something, which we should not have
|
||||
return false
|
||||
}); err != nil && !backend.IsClosed(err) {
|
||||
panic(err)
|
||||
}
|
||||
t.close()
|
||||
|
||||
return ok
|
||||
}
|
||||
|
||||
func (s *FileSet) Drop(device protocol.DeviceID) {
|
||||
@@ -150,7 +199,20 @@ func (s *FileSet) Drop(device protocol.DeviceID) {
|
||||
s.meta.resetAll(device)
|
||||
}
|
||||
|
||||
if err := s.meta.toDB(s.db, []byte(s.folder)); backend.IsClosed(err) {
|
||||
t, err := s.db.newReadWriteTransaction()
|
||||
if backend.IsClosed(err) {
|
||||
return
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer t.close()
|
||||
|
||||
if err := s.meta.toDB(t, []byte(s.folder)); backend.IsClosed(err) {
|
||||
return
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if err := t.Commit(); backend.IsClosed(err) {
|
||||
return
|
||||
} else if err != nil {
|
||||
panic(err)
|
||||
@@ -168,12 +230,6 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
|
||||
s.updateMutex.Lock()
|
||||
defer s.updateMutex.Unlock()
|
||||
|
||||
defer func() {
|
||||
if err := s.meta.toDB(s.db, []byte(s.folder)); err != nil && !backend.IsClosed(err) {
|
||||
panic(err)
|
||||
}
|
||||
}()
|
||||
|
||||
if device == protocol.LocalDeviceID {
|
||||
// For the local device we have a bunch of metadata to track.
|
||||
if err := s.db.updateLocalFiles([]byte(s.folder), fs, s.meta); err != nil && !backend.IsClosed(err) {
|
||||
|
||||
+27
-18
@@ -129,27 +129,36 @@ func (f FileInfoTruncated) FileModifiedBy() protocol.ShortID {
|
||||
}
|
||||
|
||||
func (f FileInfoTruncated) ConvertToIgnoredFileInfo(by protocol.ShortID) protocol.FileInfo {
|
||||
return protocol.FileInfo{
|
||||
Name: f.Name,
|
||||
Type: f.Type,
|
||||
ModifiedS: f.ModifiedS,
|
||||
ModifiedNs: f.ModifiedNs,
|
||||
ModifiedBy: by,
|
||||
Version: f.Version,
|
||||
RawBlockSize: f.RawBlockSize,
|
||||
LocalFlags: protocol.FlagLocalIgnored,
|
||||
}
|
||||
file := f.copyToFileInfo()
|
||||
file.SetIgnored(by)
|
||||
return file
|
||||
}
|
||||
|
||||
func (f FileInfoTruncated) ConvertToDeletedFileInfo(by protocol.ShortID, localFlags uint32) protocol.FileInfo {
|
||||
func (f FileInfoTruncated) ConvertToDeletedFileInfo(by protocol.ShortID) protocol.FileInfo {
|
||||
file := f.copyToFileInfo()
|
||||
file.SetDeleted(by)
|
||||
return file
|
||||
}
|
||||
|
||||
// copyToFileInfo just copies all members of FileInfoTruncated to protocol.FileInfo
|
||||
func (f FileInfoTruncated) copyToFileInfo() protocol.FileInfo {
|
||||
return protocol.FileInfo{
|
||||
Name: f.Name,
|
||||
Type: f.Type,
|
||||
ModifiedS: time.Now().Unix(),
|
||||
ModifiedBy: by,
|
||||
Deleted: true,
|
||||
Version: f.Version.Update(by),
|
||||
LocalFlags: localFlags,
|
||||
Name: f.Name,
|
||||
Size: f.Size,
|
||||
ModifiedS: f.ModifiedS,
|
||||
ModifiedBy: f.ModifiedBy,
|
||||
Version: f.Version,
|
||||
Sequence: f.Sequence,
|
||||
SymlinkTarget: f.SymlinkTarget,
|
||||
BlocksHash: f.BlocksHash,
|
||||
Type: f.Type,
|
||||
Permissions: f.Permissions,
|
||||
ModifiedNs: f.ModifiedNs,
|
||||
RawBlockSize: f.RawBlockSize,
|
||||
LocalFlags: f.LocalFlags,
|
||||
Deleted: f.Deleted,
|
||||
RawInvalid: f.RawInvalid,
|
||||
NoPermissions: f.NoPermissions,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+42
-37
@@ -78,30 +78,34 @@ func (t readOnlyTransaction) unmarshalTrunc(bs []byte, trunc bool) (FileIntf, er
|
||||
return tf, nil
|
||||
}
|
||||
|
||||
var tf protocol.FileInfo
|
||||
if err := tf.Unmarshal(bs); err != nil {
|
||||
var fi protocol.FileInfo
|
||||
if err := fi.Unmarshal(bs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := t.fillBlockList(&tf); err != nil {
|
||||
if err := t.fillFileInfo(&fi); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return tf, nil
|
||||
return fi, nil
|
||||
}
|
||||
|
||||
func (t readOnlyTransaction) fillBlockList(fi *protocol.FileInfo) error {
|
||||
if len(fi.BlocksHash) == 0 {
|
||||
return nil
|
||||
// fillFileInfo follows the (possible) indirection of blocks and fills it out.
|
||||
func (t readOnlyTransaction) fillFileInfo(fi *protocol.FileInfo) error {
|
||||
var key []byte
|
||||
|
||||
if len(fi.Blocks) == 0 && len(fi.BlocksHash) != 0 {
|
||||
// The blocks list is indirected and we need to load it.
|
||||
key = t.keyer.GenerateBlockListKey(key, fi.BlocksHash)
|
||||
bs, err := t.Get(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var bl BlockList
|
||||
if err := bl.Unmarshal(bs); err != nil {
|
||||
return err
|
||||
}
|
||||
fi.Blocks = bl.Blocks
|
||||
}
|
||||
blocksKey := t.keyer.GenerateBlockListKey(nil, fi.BlocksHash)
|
||||
bs, err := t.Get(blocksKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var bl BlockList
|
||||
if err := bl.Unmarshal(bs); err != nil {
|
||||
return err
|
||||
}
|
||||
fi.Blocks = bl.Blocks
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -453,26 +457,33 @@ func (t readWriteTransaction) close() {
|
||||
t.WriteTransaction.Release()
|
||||
}
|
||||
|
||||
func (t readWriteTransaction) putFile(key []byte, fi protocol.FileInfo) error {
|
||||
if fi.Blocks != nil {
|
||||
if fi.BlocksHash == nil {
|
||||
fi.BlocksHash = protocol.BlocksHash(fi.Blocks)
|
||||
}
|
||||
blocksKey := t.keyer.GenerateBlockListKey(nil, fi.BlocksHash)
|
||||
if _, err := t.Get(blocksKey); backend.IsNotFound(err) {
|
||||
func (t readWriteTransaction) putFile(fkey []byte, fi protocol.FileInfo) error {
|
||||
var bkey []byte
|
||||
|
||||
// Always set the blocks hash when there are blocks.
|
||||
if len(fi.Blocks) > 0 {
|
||||
fi.BlocksHash = protocol.BlocksHash(fi.Blocks)
|
||||
} else {
|
||||
fi.BlocksHash = nil
|
||||
}
|
||||
|
||||
// Indirect the blocks if the block list is large enough.
|
||||
if len(fi.Blocks) > blocksIndirectionCutoff {
|
||||
bkey = t.keyer.GenerateBlockListKey(bkey, fi.BlocksHash)
|
||||
if _, err := t.Get(bkey); backend.IsNotFound(err) {
|
||||
// Marshal the block list and save it
|
||||
blocksBs := mustMarshal(&BlockList{Blocks: fi.Blocks})
|
||||
if err := t.Put(blocksKey, blocksBs); err != nil {
|
||||
if err := t.Put(bkey, blocksBs); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
fi.Blocks = nil
|
||||
}
|
||||
|
||||
fi.Blocks = nil
|
||||
fiBs := mustMarshal(&fi)
|
||||
return t.Put(key, fiBs)
|
||||
return t.Put(fkey, fiBs)
|
||||
}
|
||||
|
||||
// updateGlobal adds this device+version to the version list for the given
|
||||
@@ -723,15 +734,12 @@ func (t *readWriteTransaction) withAllFolderTruncated(folder []byte, fn func(dev
|
||||
}
|
||||
continue
|
||||
}
|
||||
var f FileInfoTruncated
|
||||
// The iterator function may keep a reference to the unmarshalled
|
||||
// struct, which in turn references the buffer it was unmarshalled
|
||||
// from. dbi.Value() just returns an internal slice that it reuses, so
|
||||
// we need to copy it.
|
||||
err := f.Unmarshal(append([]byte{}, dbi.Value()...))
|
||||
|
||||
intf, err := t.unmarshalTrunc(dbi.Value(), true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
f := intf.(FileInfoTruncated)
|
||||
|
||||
switch f.Name {
|
||||
case "", ".", "..", "/": // A few obviously invalid filenames
|
||||
@@ -755,10 +763,7 @@ func (t *readWriteTransaction) withAllFolderTruncated(folder []byte, fn func(dev
|
||||
return nil
|
||||
}
|
||||
}
|
||||
if err := dbi.Error(); err != nil {
|
||||
return err
|
||||
}
|
||||
return t.Commit()
|
||||
return dbi.Error()
|
||||
}
|
||||
|
||||
type marshaller interface {
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package discover
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
stdsync "sync"
|
||||
"time"
|
||||
@@ -73,7 +74,7 @@ func (m *cachingMux) Add(finder Finder, cacheTime, negCacheTime time.Duration) {
|
||||
|
||||
// Lookup attempts to resolve the device ID using any of the added Finders,
|
||||
// while obeying the cache settings.
|
||||
func (m *cachingMux) Lookup(deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
func (m *cachingMux) Lookup(ctx context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
m.mut.RLock()
|
||||
for i, finder := range m.finders {
|
||||
if cacheEntry, ok := m.caches[i].Get(deviceID); ok {
|
||||
@@ -99,7 +100,7 @@ func (m *cachingMux) Lookup(deviceID protocol.DeviceID) (addresses []string, err
|
||||
}
|
||||
|
||||
// Perform the actual lookup and cache the result.
|
||||
if addrs, err := finder.Lookup(deviceID); err == nil {
|
||||
if addrs, err := finder.Lookup(ctx, deviceID); err == nil {
|
||||
l.Debugln("lookup for", deviceID, "at", finder)
|
||||
l.Debugln(" addresses:", addrs)
|
||||
addresses = append(addresses, addrs...)
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package discover
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -39,7 +40,9 @@ func TestCacheUnique(t *testing.T) {
|
||||
f1 := &fakeDiscovery{addresses0}
|
||||
c.Add(f1, time.Minute, 0)
|
||||
|
||||
addr, err := c.Lookup(protocol.LocalDeviceID)
|
||||
ctx := context.Background()
|
||||
|
||||
addr, err := c.Lookup(ctx, protocol.LocalDeviceID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -53,7 +56,7 @@ func TestCacheUnique(t *testing.T) {
|
||||
f2 := &fakeDiscovery{addresses1}
|
||||
c.Add(f2, time.Minute, 0)
|
||||
|
||||
addr, err = c.Lookup(protocol.LocalDeviceID)
|
||||
addr, err = c.Lookup(ctx, protocol.LocalDeviceID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -66,7 +69,7 @@ type fakeDiscovery struct {
|
||||
addresses []string
|
||||
}
|
||||
|
||||
func (f *fakeDiscovery) Lookup(deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
func (f *fakeDiscovery) Lookup(_ context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
return f.addresses, nil
|
||||
}
|
||||
|
||||
@@ -96,7 +99,7 @@ func TestCacheSlowLookup(t *testing.T) {
|
||||
// Start a lookup, which will take at least a second
|
||||
|
||||
t0 := time.Now()
|
||||
go c.Lookup(protocol.LocalDeviceID)
|
||||
go c.Lookup(context.Background(), protocol.LocalDeviceID)
|
||||
<-started // The slow lookup method has been called so we're inside the lock
|
||||
|
||||
// It should be possible to get ChildErrors while it's running
|
||||
@@ -116,7 +119,7 @@ type slowDiscovery struct {
|
||||
started chan struct{}
|
||||
}
|
||||
|
||||
func (f *slowDiscovery) Lookup(deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
func (f *slowDiscovery) Lookup(_ context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
|
||||
close(f.started)
|
||||
time.Sleep(f.delay)
|
||||
return nil, nil
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package discover
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/protocol"
|
||||
@@ -15,7 +16,7 @@ import (
|
||||
|
||||
// A Finder provides lookup services of some kind.
|
||||
type Finder interface {
|
||||
Lookup(deviceID protocol.DeviceID) (address []string, err error)
|
||||
Lookup(ctx context.Context, deviceID protocol.DeviceID) (address []string, err error)
|
||||
Error() error
|
||||
String() string
|
||||
Cache() map[protocol.DeviceID]CacheEntry
|
||||
|
||||
+44
-15
@@ -41,8 +41,8 @@ type globalClient struct {
|
||||
}
|
||||
|
||||
type httpClient interface {
|
||||
Get(url string) (*http.Response, error)
|
||||
Post(url, ctype string, data io.Reader) (*http.Response, error)
|
||||
Get(ctx context.Context, url string) (*http.Response, error)
|
||||
Post(ctx context.Context, url, ctype string, data io.Reader) (*http.Response, error)
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -89,7 +89,7 @@ func NewGlobal(server string, cert tls.Certificate, addrList AddressLister, evLo
|
||||
// The http.Client used for announcements. It needs to have our
|
||||
// certificate to prove our identity, and may or may not verify the server
|
||||
// certificate depending on the insecure setting.
|
||||
var announceClient httpClient = &http.Client{
|
||||
var announceClient httpClient = &contextClient{&http.Client{
|
||||
Timeout: requestTimeout,
|
||||
Transport: &http.Transport{
|
||||
DialContext: dialer.DialContext,
|
||||
@@ -99,14 +99,14 @@ func NewGlobal(server string, cert tls.Certificate, addrList AddressLister, evLo
|
||||
Certificates: []tls.Certificate{cert},
|
||||
},
|
||||
},
|
||||
}
|
||||
}}
|
||||
if opts.id != "" {
|
||||
announceClient = newIDCheckingHTTPClient(announceClient, devID)
|
||||
}
|
||||
|
||||
// The http.Client used for queries. We don't need to present our
|
||||
// certificate here, so lets not include it. May be insecure if requested.
|
||||
var queryClient httpClient = &http.Client{
|
||||
var queryClient httpClient = &contextClient{&http.Client{
|
||||
Timeout: requestTimeout,
|
||||
Transport: &http.Transport{
|
||||
DialContext: dialer.DialContext,
|
||||
@@ -115,7 +115,7 @@ func NewGlobal(server string, cert tls.Certificate, addrList AddressLister, evLo
|
||||
InsecureSkipVerify: opts.insecure,
|
||||
},
|
||||
},
|
||||
}
|
||||
}}
|
||||
if opts.id != "" {
|
||||
queryClient = newIDCheckingHTTPClient(queryClient, devID)
|
||||
}
|
||||
@@ -139,7 +139,7 @@ func NewGlobal(server string, cert tls.Certificate, addrList AddressLister, evLo
|
||||
}
|
||||
|
||||
// Lookup returns the list of addresses where the given device is available
|
||||
func (c *globalClient) Lookup(device protocol.DeviceID) (addresses []string, err error) {
|
||||
func (c *globalClient) Lookup(ctx context.Context, device protocol.DeviceID) (addresses []string, err error) {
|
||||
if c.noLookup {
|
||||
return nil, lookupError{
|
||||
error: errors.New("lookups not supported"),
|
||||
@@ -156,7 +156,7 @@ func (c *globalClient) Lookup(device protocol.DeviceID) (addresses []string, err
|
||||
q.Set("device", device.String())
|
||||
qURL.RawQuery = q.Encode()
|
||||
|
||||
resp, err := c.queryClient.Get(qURL.String())
|
||||
resp, err := c.queryClient.Get(ctx, qURL.String())
|
||||
if err != nil {
|
||||
l.Debugln("globalClient.Lookup", qURL, err)
|
||||
return nil, err
|
||||
@@ -211,7 +211,7 @@ func (c *globalClient) serve(ctx context.Context) {
|
||||
timer.Reset(2 * time.Second)
|
||||
|
||||
case <-timer.C:
|
||||
c.sendAnnouncement(timer)
|
||||
c.sendAnnouncement(ctx, timer)
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
@@ -219,7 +219,7 @@ func (c *globalClient) serve(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *globalClient) sendAnnouncement(timer *time.Timer) {
|
||||
func (c *globalClient) sendAnnouncement(ctx context.Context, timer *time.Timer) {
|
||||
var ann announcement
|
||||
if c.addrList != nil {
|
||||
ann.Addresses = c.addrList.ExternalAddresses()
|
||||
@@ -239,7 +239,7 @@ func (c *globalClient) sendAnnouncement(timer *time.Timer) {
|
||||
|
||||
l.Debugf("Announcement: %s", postData)
|
||||
|
||||
resp, err := c.announceClient.Post(c.server, "application/json", bytes.NewReader(postData))
|
||||
resp, err := c.announceClient.Post(ctx, c.server, "application/json", bytes.NewReader(postData))
|
||||
if err != nil {
|
||||
l.Debugln("announce POST:", err)
|
||||
c.setError(err)
|
||||
@@ -362,8 +362,8 @@ func (c *idCheckingHTTPClient) check(resp *http.Response) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *idCheckingHTTPClient) Get(url string) (*http.Response, error) {
|
||||
resp, err := c.httpClient.Get(url)
|
||||
func (c *idCheckingHTTPClient) Get(ctx context.Context, url string) (*http.Response, error) {
|
||||
resp, err := c.httpClient.Get(ctx, url)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -374,8 +374,8 @@ func (c *idCheckingHTTPClient) Get(url string) (*http.Response, error) {
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (c *idCheckingHTTPClient) Post(url, ctype string, data io.Reader) (*http.Response, error) {
|
||||
resp, err := c.httpClient.Post(url, ctype, data)
|
||||
func (c *idCheckingHTTPClient) Post(ctx context.Context, url, ctype string, data io.Reader) (*http.Response, error) {
|
||||
resp, err := c.httpClient.Post(ctx, url, ctype, data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -403,3 +403,32 @@ func (e *errorHolder) Error() error {
|
||||
e.mut.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
type contextClient struct {
|
||||
*http.Client
|
||||
}
|
||||
|
||||
func (c *contextClient) Get(ctx context.Context, url string) (*http.Response, error) {
|
||||
// For <go1.13 compatibility. Use the following commented line once that
|
||||
// isn't required anymore.
|
||||
// req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
|
||||
req, err := http.NewRequest("GET", url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Cancel = ctx.Done()
|
||||
return c.Client.Do(req)
|
||||
}
|
||||
|
||||
func (c *contextClient) Post(ctx context.Context, url, ctype string, data io.Reader) (*http.Response, error) {
|
||||
// For <go1.13 compatibility. Use the following commented line once that
|
||||
// isn't required anymore.
|
||||
// req, err := http.NewRequestWithContext(ctx, "POST", url, data)
|
||||
req, err := http.NewRequest("POST", url, data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Cancel = ctx.Done()
|
||||
req.Header.Set("Content-Type", ctype)
|
||||
return c.Client.Do(req)
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
package discover
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
@@ -225,7 +226,7 @@ func testLookup(url string) ([]string, error) {
|
||||
go disco.Serve()
|
||||
defer disco.Stop()
|
||||
|
||||
return disco.Lookup(protocol.LocalDeviceID)
|
||||
return disco.Lookup(context.Background(), protocol.LocalDeviceID)
|
||||
}
|
||||
|
||||
type fakeDiscoveryServer struct {
|
||||
|
||||
@@ -91,7 +91,7 @@ func NewLocal(id protocol.DeviceID, addr string, addrList AddressLister, evLogge
|
||||
}
|
||||
|
||||
// Lookup returns a list of addresses the device is available at.
|
||||
func (c *localClient) Lookup(device protocol.DeviceID) (addresses []string, err error) {
|
||||
func (c *localClient) Lookup(_ context.Context, device protocol.DeviceID) (addresses []string, err error) {
|
||||
if cache, ok := c.Get(device); ok {
|
||||
if time.Since(cache.when) < CacheLifeTime {
|
||||
addresses = cache.Addresses
|
||||
|
||||
+2
-1
@@ -532,7 +532,8 @@ func (f *folder) scanSubdirs(subDirs []string) error {
|
||||
}
|
||||
return true
|
||||
}
|
||||
nf := file.ConvertToDeletedFileInfo(f.shortID, f.localFlags)
|
||||
nf := file.ConvertToDeletedFileInfo(f.shortID)
|
||||
nf.LocalFlags = f.localFlags
|
||||
if file.ShouldConflict() {
|
||||
// We do not want to override the global version with
|
||||
// the deleted file. Setting to an empty version makes
|
||||
|
||||
@@ -104,14 +104,8 @@ func (f *receiveOnlyFolder) Revert() {
|
||||
return true // continue
|
||||
}
|
||||
|
||||
fi = protocol.FileInfo{
|
||||
Name: fi.Name,
|
||||
Type: fi.Type,
|
||||
ModifiedS: time.Now().Unix(),
|
||||
ModifiedBy: f.shortID,
|
||||
Deleted: true,
|
||||
Version: protocol.Vector{}, // if this file ever resurfaces anywhere we want our delete to be strictly older
|
||||
}
|
||||
fi.SetDeleted(f.shortID)
|
||||
fi.Version = protocol.Vector{} // if this file ever resurfaces anywhere we want our delete to be strictly older
|
||||
} else {
|
||||
// Revert means to throw away our local changes. We reset the
|
||||
// version to the empty vector, which is strictly older than any
|
||||
|
||||
@@ -118,10 +118,7 @@ func (f *sendOnlyFolder) Override() {
|
||||
}
|
||||
if !ok || have.Name != need.Name {
|
||||
// We are missing the file
|
||||
need.Deleted = true
|
||||
need.Blocks = nil
|
||||
need.Version = need.Version.Update(f.shortID)
|
||||
need.Size = 0
|
||||
need.SetDeleted(f.shortID)
|
||||
} else {
|
||||
// We have the file, replace with our version
|
||||
have.Version = have.Version.Merge(need.Version).Update(f.shortID)
|
||||
|
||||
@@ -270,7 +270,12 @@ func (c *folderSummaryService) calculateSummaries(ctx context.Context) {
|
||||
case <-pump.C:
|
||||
t0 := time.Now()
|
||||
for _, folder := range c.foldersToHandle() {
|
||||
c.sendSummary(folder)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
c.sendSummary(ctx, folder)
|
||||
}
|
||||
|
||||
// We don't want to spend all our time calculating summaries. Lets
|
||||
@@ -280,7 +285,7 @@ func (c *folderSummaryService) calculateSummaries(ctx context.Context) {
|
||||
pump.Reset(wait)
|
||||
|
||||
case folder := <-c.immediate:
|
||||
c.sendSummary(folder)
|
||||
c.sendSummary(ctx, folder)
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
@@ -313,7 +318,7 @@ func (c *folderSummaryService) foldersToHandle() []string {
|
||||
}
|
||||
|
||||
// sendSummary send the summary events for a single folder
|
||||
func (c *folderSummaryService) sendSummary(folder string) {
|
||||
func (c *folderSummaryService) sendSummary(ctx context.Context, folder string) {
|
||||
// The folder summary contains how many bytes, files etc
|
||||
// are in the folder and how in sync we are.
|
||||
data, err := c.Summary(folder)
|
||||
@@ -326,6 +331,12 @@ func (c *folderSummaryService) sendSummary(folder string) {
|
||||
})
|
||||
|
||||
for _, devCfg := range c.cfg.Folders()[folder].Devices {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
if devCfg.DeviceID.Equals(c.id) {
|
||||
// We already know about ourselves.
|
||||
continue
|
||||
|
||||
@@ -266,24 +266,36 @@ func BlocksEqual(a, b []BlockInfo) bool {
|
||||
}
|
||||
|
||||
func (f *FileInfo) SetMustRescan(by ShortID) {
|
||||
f.LocalFlags = FlagLocalMustRescan
|
||||
f.ModifiedBy = by
|
||||
f.Blocks = nil
|
||||
f.Sequence = 0
|
||||
f.setLocalFlags(by, FlagLocalMustRescan)
|
||||
}
|
||||
|
||||
func (f *FileInfo) SetIgnored(by ShortID) {
|
||||
f.LocalFlags = FlagLocalIgnored
|
||||
f.ModifiedBy = by
|
||||
f.Blocks = nil
|
||||
f.Sequence = 0
|
||||
f.setLocalFlags(by, FlagLocalIgnored)
|
||||
}
|
||||
|
||||
func (f *FileInfo) SetUnsupported(by ShortID) {
|
||||
f.LocalFlags = FlagLocalUnsupported
|
||||
f.setLocalFlags(by, FlagLocalUnsupported)
|
||||
}
|
||||
|
||||
func (f *FileInfo) SetDeleted(by ShortID) {
|
||||
f.ModifiedBy = by
|
||||
f.Deleted = true
|
||||
f.Version = f.Version.Update(by)
|
||||
f.ModifiedS = time.Now().Unix()
|
||||
f.setNoContent()
|
||||
}
|
||||
|
||||
func (f *FileInfo) setLocalFlags(by ShortID, flags uint32) {
|
||||
f.RawInvalid = false
|
||||
f.LocalFlags = flags
|
||||
f.ModifiedBy = by
|
||||
f.setNoContent()
|
||||
}
|
||||
|
||||
func (f *FileInfo) setNoContent() {
|
||||
f.Blocks = nil
|
||||
f.Sequence = 0
|
||||
f.BlocksHash = nil
|
||||
f.Size = 0
|
||||
}
|
||||
|
||||
func (b BlockInfo) String() string {
|
||||
|
||||
+24
-13
@@ -160,12 +160,6 @@ USER-AGENT: syncthing/1.0
|
||||
}
|
||||
defer socket.Close() // Make sure our socket gets closed
|
||||
|
||||
err = socket.SetDeadline(time.Now().Add(timeout))
|
||||
if err != nil {
|
||||
l.Debugln("UPnP discovery: setting socket deadline:", err)
|
||||
return
|
||||
}
|
||||
|
||||
l.Debugln("Sending search request for device type", deviceType, "on", intf.Name)
|
||||
|
||||
_, err = socket.WriteTo(search, ssdp)
|
||||
@@ -178,16 +172,33 @@ USER-AGENT: syncthing/1.0
|
||||
|
||||
l.Debugln("Listening for UPnP response for device type", deviceType, "on", intf.Name)
|
||||
|
||||
// Listen for responses until a timeout is reached
|
||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
// Listen for responses until a timeout is reached or the context is
|
||||
// cancelled
|
||||
resp := make([]byte, 65536)
|
||||
loop:
|
||||
for {
|
||||
resp := make([]byte, 65536)
|
||||
n, _, err := socket.ReadFrom(resp)
|
||||
if err != nil {
|
||||
if e, ok := err.(net.Error); !ok || !e.Timeout() {
|
||||
l.Infoln("UPnP read:", err) //legitimate error, not a timeout.
|
||||
}
|
||||
if err := socket.SetDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
|
||||
l.Infoln("UPnP socket:", err)
|
||||
break
|
||||
}
|
||||
|
||||
n, _, err := socket.ReadFrom(resp)
|
||||
if err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break loop
|
||||
default:
|
||||
}
|
||||
if e, ok := err.(net.Error); ok && e.Timeout() {
|
||||
continue // continue reading
|
||||
}
|
||||
l.Infoln("UPnP read:", err) //legitimate error, not a timeout.
|
||||
break
|
||||
}
|
||||
|
||||
igds, err := parseResponse(ctx, deviceType, resp[:n])
|
||||
if err != nil {
|
||||
switch err.(type) {
|
||||
|
||||
Reference in New Issue
Block a user