ronantakizawa/codereview-bench huggingface dataset:
Repository: syncthing/syncthing
File: lib/model/folder_sendrecv.go (Go)

Code before:

		l.Debugln(f, "Deleting file", file.Name)
		if update, err := f.deleteFile(file, scanChan); err != nil {
			f.newPullError(file.Name, errors.Wrap(err, "delete file"))
		} else {
			dbUpdateChan <- update
		}
	}

	for i := range dirDeletions {
		select {
		case <-f.ctx.Done():
			return
		default:
		}

		dir := dirDeletions[len(dirDeletions)-i-1]
		l.Debugln(f, "Deleting dir", dir.Name)
		f.deleteDir(dir, dbUpdateChan, scanChan)
	}
}

// handleDir creates or updates the given directory
func (f *sendReceiveFolder) handleDir(file protocol.FileInfo, dbUpdateChan chan<- dbUpdateJob, scanChan chan<- string) {
	// Used in the defer closure below, updated by the function body. Take
	// care not declare another err.
	var err error

	events.Default.Log(events.ItemStarted, map[string]string{
		"folder": f.folderID,
		"item":   file.Name,
		"type":   "dir",
		"action": "update",
	})

	defer func() {
		events.Default.Log(events.ItemFinished, map[string]interface{}{
			"folder": f.folderID,
			"item":   file.Name,
			"error":  events.Error(err),
			"type":   "dir",
			"action": "update",
		})
	}()

	mode := fs.FileMode(file.Permissions & 0777)
	if f.IgnorePerms || file.NoPermissions {
		mode = 0777
	}

	if shouldDebug() {

Review comment (bug):
Now we try to delete `a\b` before `a\b\c`. Seems wrong?

Code after:

func (f *sendReceiveFolder) processDeletions(fileDeletions map[string]protocol.FileInfo, dirDeletions []protocol.FileInfo, dbUpdateChan chan<- dbUpdateJob, scanChan chan<- string) {
	for _, file := range fileDeletions {
		select {
		case <-f.ctx.Done():
			return
		default:
		}

		l.Debugln(f, "Deleting file", file.Name)
		f.resetPullError(file.Name)
		if update, err := f.deleteFile(file, scanChan); err != nil {
			f.newPullError(file.Name, errors.Wrap(err, "delete file"))
		} else {
			dbUpdateChan <- update
		}
	}

	for _, dir := range dirDeletions {
		select {
		case <-f.ctx.Done():
			return
		default:
		}

		f.resetPullError(dir.Name)
		l.Debugln(f, "Deleting dir", dir.Name)
		f.deleteDir(dir, dbUpdateChan, scanChan)
	}
}

// handleDir creates or updates the given directory
func (f *sendReceiveFolder) handleDir(file protocol.FileInfo, dbUpdateChan chan<- dbUpdateJob, scanChan chan<- string) {
	// Used in the defer closure below, updated by the function body. Take
	// care not declare another err.
	var err error

	events.Default.Log(events.ItemStarted, map[string]string{
		"folder": f.folderID,
		"item":   file.Name,
		"type":   "dir",
		"action": "update",
	})

	defer func() {
		events.Default.Log(events.ItemFinished, map[string]interface{}{
			"folder": f.folderID,
			"item":   file.Name,
			"error":  events.Error(err),
			"type":   "dir",
			"action": "update",
Repository: syncthing/syncthing
File: lib/model/model_test.go (Go)

Code before:
	default:
	}
	first.Close()
	select {
	case <-returned:
	case <-time.After(time.Second):
		t.Fatalf("Second request did not return after first was done")
	}
}

func TestSanitizePath(t *testing.T) {
	cases := [][2]string{
		{"", ""},
		{"foo", "foo"},
		{`\*/foo\?/bar[{!@$%^&*#()}]`, "foo bar ()"},
		{"Räksmörgås", "Räksmörgås"},
		{`Räk \/ smörgås`, "Räk smörgås"},
		{"هذا هو *\x07?اسم الملف", "هذا هو اسم الملف"},
		{`../foo.txt`, `.. foo.txt`},
	}

	for _, tc := range cases {
		res := sanitizePath(tc[0])
		if res != tc[1] {
			t.Errorf("sanitizePath(%q) => %q, expected %q", tc[0], res, tc[1])
		}
	}
}

// TestConnCloseOnRestart checks that there is no deadlock when calling Close
// on a protocol connection that has a blocking reader (blocking writer can't
// be done as the test requires clusterconfigs to go through).
func TestConnCloseOnRestart(t *testing.T) {
	w, fcfg := tmpDefaultWrapper()
	m := setupModel(w)
	defer cleanupModelAndRemoveDir(m, fcfg.Filesystem().URI())

	br := &testutils.BlockingRW{}
	nw := &testutils.NoopRW{}
	m.AddConnection(newFakeProtoConn(protocol.NewConnection(device1, br, nw, m, "testConn", protocol.CompressNever)), protocol.HelloResult{})

	newFcfg := fcfg.Copy()
	newFcfg.Paused = true
	done := make(chan struct{})
	go func() {
		m.RestartFolder(fcfg, newFcfg)
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):

Review comment (suggestion):
Not relevant to this, but we could just get all the folders from the config (that exits on the mode) and remove them, without having to pass a URI.

Code after:
func TestSanitizePath(t *testing.T) {
	cases := [][2]string{
		{"", ""},
		{"foo", "foo"},
		{`\*/foo\?/bar[{!@$%^&*#()}]`, "foo bar ()"},
		{"Räksmörgås", "Räksmörgås"},
		{`Räk \/ smörgås`, "Räk smörgås"},
		{"هذا هو *\x07?اسم الملف", "هذا هو اسم الملف"},
		{`../foo.txt`, `.. foo.txt`},
	}

	for _, tc := range cases {
		res := sanitizePath(tc[0])
		if res != tc[1] {
			t.Errorf("sanitizePath(%q) => %q, expected %q", tc[0], res, tc[1])
		}
	}
}

// TestConnCloseOnRestart checks that there is no deadlock when calling Close
// on a protocol connection that has a blocking reader (blocking writer can't
// be done as the test requires clusterconfigs to go through).
func TestConnCloseOnRestart(t *testing.T) {
	w, fcfg := tmpDefaultWrapper()
	m := setupModel(w)
	defer cleanupModelAndRemoveDir(m, fcfg.Filesystem().URI())

	br := &testutils.BlockingRW{}
	nw := &testutils.NoopRW{}
	m.AddConnection(newFakeProtoConn(protocol.NewConnection(device1, br, nw, m, "testConn", protocol.CompressNever)), protocol.HelloResult{})

	newFcfg := fcfg.Copy()
	newFcfg.Paused = true
	done := make(chan struct{})
	go func() {
		m.RestartFolder(fcfg, newFcfg)
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("Timed out before folder restart returned")
	}
	m.pmut.RLock()
	defer m.pmut.RUnlock()
	if len(m.conn) != 0 {
		t.Errorf("Conn wasn't removed on restart (len(m.conn) == %v)", len(m.conn))
	}
}
Repository: syncthing/syncthing
File: lib/connections/tcp_listen.go (Go)

Code before:
	util.ServiceWithError
	onAddressesChangedNotifier

	uri     *url.URL
	cfg     config.Wrapper
	tlsCfg  *tls.Config
	conns   chan internalConn
	factory listenerFactory

	natService *nat.Service
	mapping    *nat.Mapping

	mut sync.RWMutex
}

func (t *tcpListener) serve(stop chan struct{}) error {
	tcaddr, err := net.ResolveTCPAddr(t.uri.Scheme, t.uri.Host)
	if err != nil {
		l.Infoln("Listen (BEP/tcp):", err)
		return err
	}

	lc := net.ListenConfig{
		Control: dialer.ReusePortControl,
	}

	listener, err := lc.Listen(context.TODO(), t.uri.Scheme, tcaddr.String())
	if err != nil {
		l.Infoln("Listen (BEP/tcp):", err)
		return err
	}
	registry.Register(t.uri.Scheme, tcaddr)

	defer listener.Close()
	defer registry.Unregister(t.uri.Scheme, tcaddr)

	l.Infof("TCP listener (%v) starting", listener.Addr())
	defer l.Infof("TCP listener (%v) shutting down", listener.Addr())

	mapping := t.natService.NewMapping(nat.TCP, tcaddr.IP, tcaddr.Port)
	mapping.OnChanged(func(_ *nat.Mapping, _, _ []nat.Address) {
		t.notifyAddressesChanged(t)
	})
	defer t.natService.RemoveMapping(mapping)

	t.mut.Lock()
	t.mapping = mapping
	t.mut.Unlock()

	acceptFailures := 0
	const maxAcceptFailures = 10

Review comment (suggestion):
`lc` is only used in the line below, therefore I wouldn't separate it.

Code after:
	util.ServiceWithError
	onAddressesChangedNotifier

	uri     *url.URL
	cfg     config.Wrapper
	tlsCfg  *tls.Config
	conns   chan internalConn
	factory listenerFactory

	natService *nat.Service
	mapping    *nat.Mapping

	mut sync.RWMutex
}

func (t *tcpListener) serve(ctx context.Context) error {
	tcaddr, err := net.ResolveTCPAddr(t.uri.Scheme, t.uri.Host)
	if err != nil {
		l.Infoln("Listen (BEP/tcp):", err)
		return err
	}

	lc := net.ListenConfig{
		Control: dialer.ReusePortControl,
	}

	listener, err := lc.Listen(context.TODO(), t.uri.Scheme, tcaddr.String())
	if err != nil {
		l.Infoln("Listen (BEP/tcp):", err)
		return err
	}
	t.notifyAddressesChanged(t)
	registry.Register(t.uri.Scheme, tcaddr)

	defer listener.Close()
	defer t.clearAddresses(t)
	defer registry.Unregister(t.uri.Scheme, tcaddr)

	l.Infof("TCP listener (%v) starting", listener.Addr())
	defer l.Infof("TCP listener (%v) shutting down", listener.Addr())

	mapping := t.natService.NewMapping(nat.TCP, tcaddr.IP, tcaddr.Port)
	mapping.OnChanged(func(_ *nat.Mapping, _, _ []nat.Address) {
		t.notifyAddressesChanged(t)
	})
	defer t.natService.RemoveMapping(mapping)

	t.mut.Lock()
	t.mapping = mapping
	t.mut.Unlock()
Repository: syncthing/syncthing
File: lib/dialer/control_unix.go (Go)

Code before:
	"golang.org/x/sys/unix"
)

var SupportsReusePort = false

func init() {
	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM, unix.IPPROTO_IP)
	if err != nil {
		l.Debugln("Failed to create a socket", err)
		return
	}
	defer func() { _ = unix.Close(fd) }()

	err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
	switch {
	case err == unix.ENOPROTOOPT || err == unix.EINVAL:
		l.Debugln("SO_REUSEPORT not supported")
	case err != nil:
		l.Debugln("Unknown error when determining SO_REUSEPORT support", err)
	default:
		l.Debugln("SO_REUSEPORT supported")
		SupportsReusePort = true
	}
}

func ReusePortControl(network, address string, c syscall.RawConn) error {
	if !SupportsReusePort {
		return nil
	}
	var opErr error
	err := c.Control(func(fd uintptr) {
		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
	})
	if err != nil {
		return err
	}
	return opErr
}


Review comment (suggestion):
`network` and `address` could be `_` to indicate that their unused.

Code after:
	"golang.org/x/sys/unix"
)

var SupportsReusePort = false

func init() {
	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM, unix.IPPROTO_IP)
	if err != nil {
		l.Debugln("Failed to create a socket", err)
		return
	}
	defer func() { _ = unix.Close(fd) }()

	err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
	switch {
	case err == unix.ENOPROTOOPT || err == unix.EINVAL:
		l.Debugln("SO_REUSEPORT not supported")
	case err != nil:
		l.Debugln("Unknown error when determining SO_REUSEPORT support", err)
	default:
		l.Debugln("SO_REUSEPORT supported")
		SupportsReusePort = true
	}
}

func ReusePortControl(_, _ string, c syscall.RawConn) error {
	if !SupportsReusePort {
		return nil
	}
	var opErr error
	err := c.Control(func(fd uintptr) {
		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
	})
	if err != nil {
		return err
	}
	return opErr
}
Repository: syncthing/syncthing
File: lib/connections/quic_listen.go (Go)

Code before:

	onAddressesChangedNotifier

	uri     *url.URL
	cfg     config.Wrapper
	tlsCfg  *tls.Config
	stop    chan struct{}
	conns   chan internalConn
	factory listenerFactory

	address *url.URL
	err     error
	mut     sync.Mutex
}

func (t *quicListener) OnNATTypeChanged(natType stun.NATType) {
	if natType != stun.NATUnknown {
		l.Infof("%s detected NAT type: %s", t.uri, natType)
	}
	t.nat.Store(natType)
}

func (t *quicListener) OnExternalAddressChanged(address *stun.Host, via string) {
	var uri *url.URL
	if address != nil {
		uri = &url.URL{}
		*uri = *t.uri
		uri.Host = address.TransportAddr()
	}

	t.mut.Lock()
	existingAddress := t.address
	t.address = uri
	t.mut.Unlock()

	if uri != nil && (existingAddress == nil || existingAddress.String() != uri.String()) {
		l.Infof("%s resolved external address %s (via %s)", t.uri, uri.String(), via)
		t.notifyAddressesChanged(t)
	} else if uri == nil && existingAddress != nil {
		t.notifyAddressesChanged(t)
	}
}

func (t *quicListener) Serve() {
	t.mut.Lock()
	t.err = nil
	t.mut.Unlock()

	network := strings.Replace(t.uri.Scheme, "quic", "udp", -1)

	packetConn, err := net.ListenPacket(network, t.uri.Host)

Review comment (suggestion):
I'd have preferred this as

```
copy := *t.uri
uri = &copy
```

to be more explicit about it being a copy that needs to happen, but /shrug

Code after:

	onAddressesChangedNotifier

	uri     *url.URL
	cfg     config.Wrapper
	tlsCfg  *tls.Config
	stop    chan struct{}
	conns   chan internalConn
	factory listenerFactory

	address *url.URL
	err     error
	mut     sync.Mutex
}

func (t *quicListener) OnNATTypeChanged(natType stun.NATType) {
	if natType != stun.NATUnknown {
		l.Infof("%s detected NAT type: %s", t.uri, natType)
	}
	t.nat.Store(natType)
}

func (t *quicListener) OnExternalAddressChanged(address *stun.Host, via string) {
	var uri *url.URL
	if address != nil {
		copy := *t.uri
		uri = &copy
		uri.Host = address.TransportAddr()
	}

	t.mut.Lock()
	existingAddress := t.address
	t.address = uri
	t.mut.Unlock()

	if uri != nil && (existingAddress == nil || existingAddress.String() != uri.String()) {
		l.Infof("%s resolved external address %s (via %s)", t.uri, uri.String(), via)
		t.notifyAddressesChanged(t)
	} else if uri == nil && existingAddress != nil {
		t.notifyAddressesChanged(t)
	}
}

func (t *quicListener) Serve() {
	t.mut.Lock()
	t.err = nil
	t.mut.Unlock()

	network := strings.Replace(t.uri.Scheme, "quic", "udp", -1)

	packetConn, err := net.ListenPacket(network, t.uri.Host)
Repository: syncthing/syncthing
File: lib/db/lowlevel.go (Go)

Code before:

// ListFolders returns the list of folders currently in the database
func (db *Lowlevel) ListFolders() []string {
	return db.folderIdx.Values()
}

// Committed returns the number of items committed to the database since startup
func (db *Lowlevel) Committed() int64 {
	return atomic.LoadInt64(&db.committed)
}

func (db *Lowlevel) Put(key, val []byte, wo *opt.WriteOptions) error {
	atomic.AddInt64(&db.committed, 1)
	return db.DB.Put(key, val, wo)
}

func (db *Lowlevel) Delete(key []byte, wo *opt.WriteOptions) error {
	atomic.AddInt64(&db.committed, 1)
	return db.DB.Delete(key, wo)
}

func (db *Lowlevel) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.Iterator {
	return db.newIterator(func() iterator.Iterator { return db.DB.NewIterator(slice, ro) })
}

func (db *Lowlevel) newIterator(fn func() iterator.Iterator) iterator.Iterator {
	db.closeMut.RLock()
	defer db.closeMut.RUnlock()
	if db.closed {
		return &closedIter{}
	}
	db.iterWG.Add(1)
	return &iter{
		Iterator: fn(),
		db:       db,
	}
}

func (db *Lowlevel) GetSnapshot() snapshot {
	s, err := db.DB.GetSnapshot()
	if err != nil {
		if err == leveldb.ErrClosed {
			return &closedSnap{}
		}
		panic(err)
	}
	return &snap{
		Snapshot: s,
		db:       db,
	}
}

Review comment (suggestion):
comment here on why it takes a constructor function (and maybe call it something like that) instead of just an iterator, as it took me a while to figure out what was going on here and why :)

Code after:

// ListFolders returns the list of folders currently in the database
func (db *Lowlevel) ListFolders() []string {
	return db.folderIdx.Values()
}

// Committed returns the number of items committed to the database since startup
func (db *Lowlevel) Committed() int64 {
	return atomic.LoadInt64(&db.committed)
}

func (db *Lowlevel) Put(key, val []byte, wo *opt.WriteOptions) error {
	db.closeMut.RLock()
	defer db.closeMut.RUnlock()
	if db.closed {
		return leveldb.ErrClosed
	}
	atomic.AddInt64(&db.committed, 1)
	return db.DB.Put(key, val, wo)
}

func (db *Lowlevel) Write(batch *leveldb.Batch, wo *opt.WriteOptions) error {
	db.closeMut.RLock()
	defer db.closeMut.RUnlock()
	if db.closed {
		return leveldb.ErrClosed
	}
	return db.DB.Write(batch, wo)
}

func (db *Lowlevel) Delete(key []byte, wo *opt.WriteOptions) error {
	db.closeMut.RLock()
	defer db.closeMut.RUnlock()
	if db.closed {
		return leveldb.ErrClosed
	}
	atomic.AddInt64(&db.committed, 1)
	return db.DB.Delete(key, wo)
}

func (db *Lowlevel) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.Iterator {
	return db.newIterator(func() iterator.Iterator { return db.DB.NewIterator(slice, ro) })
}

// newIterator returns an iterator created with the given constructor only if db
// is not yet closed. If it is closed, a closedIter is returned instead.
func (db *Lowlevel) newIterator(constr func() iterator.Iterator) iterator.Iterator {
	db.closeMut.RLock()
	defer db.closeMut.RUnlock()
	if db.closed {
		return &closedIter{}
Repository: syncthing/syncthing
File: lib/protocol/protocol_test.go (Go)

Code before:
}

// TestClusterConfigAfterClose checks that ClusterConfig does not deadlock when
// ClusterConfig is called on a closed connection.
func TestClusterConfigAfterClose(t *testing.T) {
	m := newTestModel()

	c := NewConnection(c0ID, &testutils.BlockingRW{}, &testutils.BlockingRW{}, m, "name", CompressAlways).(wireFormatConnection).Connection.(*rawConnection)
	c.Start()

	c.internalClose(errManual)

	done := make(chan struct{})
	go func() {
		c.ClusterConfig(ClusterConfig{})
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("timed out before Cluster Config returned")
	}
}

func TestDispatcherToCloseDeadlock(t *testing.T) {
	m := newTestModel()
	c := NewConnection(c0ID, &testutils.BlockingRW{}, &testutils.NoopRW{}, m, "name", CompressAlways).(wireFormatConnection).Connection.(*rawConnection)
	m.ccFn = func(devID DeviceID, cc ClusterConfig) {
		c.Close(errManual)
	}
	c.Start()

	c.inbox <- &ClusterConfig{}

	select {
	case <-c.dispatcherLoopStopped:
	case <-time.After(time.Second):
		t.Fatal("timed out before dispatcher loop terminated")
	}
}


Review comment (bug):
// Verify that we don't deadlock when calling Close() from within one of the model callbacks (ClusterConfig).

Code after:
}

// TestClusterConfigAfterClose checks that ClusterConfig does not deadlock when
// ClusterConfig is called on a closed connection.
func TestClusterConfigAfterClose(t *testing.T) {
	m := newTestModel()

	c := NewConnection(c0ID, &testutils.BlockingRW{}, &testutils.BlockingRW{}, m, "name", CompressAlways).(wireFormatConnection).Connection.(*rawConnection)
	c.Start()

	c.internalClose(errManual)

	done := make(chan struct{})
	go func() {
		c.ClusterConfig(ClusterConfig{})
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("timed out before Cluster Config returned")
	}
}

func TestDispatcherToCloseDeadlock(t *testing.T) {
	// Verify that we don't deadlock when calling Close() from within one of
	// the model callbacks (ClusterConfig).
	m := newTestModel()
	c := NewConnection(c0ID, &testutils.BlockingRW{}, &testutils.NoopRW{}, m, "name", CompressAlways).(wireFormatConnection).Connection.(*rawConnection)
	m.ccFn = func(devID DeviceID, cc ClusterConfig) {
		c.Close(errManual)
	}
	c.Start()

	c.inbox <- &ClusterConfig{}

	select {
	case <-c.dispatcherLoopStopped:
	case <-time.After(time.Second):
		t.Fatal("timed out before dispatcher loop terminated")
	}
}
Repository: syncthing/syncthing
File: lib/model/folder.go (Go)

Code before:
	var eventChan <-chan fs.Event
	var errChan <-chan error
	warnedOutside := false
	for {
		select {
		case <-failTimer.C:
			eventChan, errChan, err = f.Filesystem().Watch(".", f.ignores, ctx, f.IgnorePerms)
			// We do this at most once per minute which is the
			// default rescan time without watcher.
			f.scanOnWatchErr()
			f.setWatchError(err)
			if err != nil {
				failTimer.Reset(time.Minute)
				continue
			}
			watchaggregator.Aggregate(eventChan, f.watchChan, f.FolderConfiguration, f.model.cfg, aggrCtx)
			l.Debugln("Started filesystem watcher for folder", f.Description())
		case err = <-errChan:
			f.setWatchError(err)
			// This error was previously a panic and should never occur, so generate
			// a warning, but don't do it repetitively.
			if !warnedOutside {
				if _, ok := err.(*fs.ErrWatchEventOutsideRoot); ok {
					l.Warnln(err)
					warnedOutside = true
					return
				}
			}
			aggrCancel()
			errChan = nil
			aggrCtx, aggrCancel = context.WithCancel(ctx)
			failTimer.Reset(time.Minute)
		case <-ctx.Done():
			return
		}
	}
}

// setWatchError sets the current error state of the watch and should be called
// regardless of whether err is nil or not.
func (f *folder) setWatchError(err error) {
	f.watchMut.Lock()
	prevErr := f.watchErr
	f.watchErr = err
	f.watchMut.Unlock()
	if err != prevErr {
		data := map[string]interface{}{
			"folder": f.ID,
		}
		if prevErr != nil {
			data["from"] = prevErr.Error()

Review comment (bug):
Returning here is clearly wrong: We still need to stop the watch, and retry (otherwise what's the point of `warnedOutside`). The bug is still minor as it only comes up when the watcher reports some event outside of the folder.

Code after:
			}
			watchaggregator.Aggregate(eventChan, f.watchChan, f.FolderConfiguration, f.model.cfg, aggrCtx)
			l.Debugln("Started filesystem watcher for folder", f.Description())
		case err = <-errChan:
			f.setWatchError(err)
			// This error was previously a panic and should never occur, so generate
			// a warning, but don't do it repetitively.
			if !warnedOutside {
				if _, ok := err.(*fs.ErrWatchEventOutsideRoot); ok {
					l.Warnln(err)
					warnedOutside = true
				}
			}
			aggrCancel()
			errChan = nil
			aggrCtx, aggrCancel = context.WithCancel(ctx)
			failTimer.Reset(time.Minute)
		case <-ctx.Done():
			return
		}
	}
}

// setWatchError sets the current error state of the watch and should be called
// regardless of whether err is nil or not.
func (f *folder) setWatchError(err error) {
	f.watchMut.Lock()
	prevErr := f.watchErr
	f.watchErr = err
	f.watchMut.Unlock()
	if err != prevErr {
		data := map[string]interface{}{
			"folder": f.ID,
		}
		if prevErr != nil {
			data["from"] = prevErr.Error()
		}
		if err != nil {
			data["to"] = err.Error()
		}
		events.Default.Log(events.FolderWatchStateChanged, data)
	}
	if err == nil {
		return
	}
	msg := fmt.Sprintf("Error while trying to start filesystem watcher for folder %s, trying again in 1min: %v", f.Description(), err)
	if prevErr != err {
		l.Infof(msg)
		return
	}
	l.Debugf(msg)