a728743c86
The readWriteTransaction offered both commit() (the one to use) and Commit() (via embedding) where the latter didn't close the read transaction. This removes the lower cased variant in order to prevent the mistake. The only place where the mistake was made was the new gc runner, where it would leave a read snapshot open forever.
614 lines
15 KiB
Go
614 lines
15 KiB
Go
// Copyright (C) 2014 The Syncthing Authors.
|
|
//
|
|
// This Source Code Form is subject to the terms of the Mozilla Public
|
|
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
|
// You can obtain one at https://mozilla.org/MPL/2.0/.
|
|
|
|
package db
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/binary"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/syncthing/syncthing/lib/db/backend"
|
|
"github.com/syncthing/syncthing/lib/protocol"
|
|
"github.com/syncthing/syncthing/lib/sync"
|
|
"github.com/willf/bloom"
|
|
)
|
|
|
|
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"
|
|
)
|
|
|
|
var blockGCInterval = blockGCDefaultInterval
|
|
|
|
func init() {
|
|
if dur, err := time.ParseDuration(os.Getenv("STGCBLOCKSEVERY")); err == nil {
|
|
blockGCInterval = dur
|
|
}
|
|
}
|
|
|
|
// Lowlevel is the lowest level database interface. It has a very simple
|
|
// purpose: hold the actual backend database, and the in-memory state
|
|
// that belong to that database. In the same way that a single on disk
|
|
// database can only be opened once, there should be only one Lowlevel for
|
|
// any given backend.
|
|
type Lowlevel struct {
|
|
backend.Backend
|
|
folderIdx *smallIndex
|
|
deviceIdx *smallIndex
|
|
keyer keyer
|
|
gcMut sync.RWMutex
|
|
gcKeyCount int
|
|
gcStop chan struct{}
|
|
}
|
|
|
|
func NewLowlevel(backend backend.Backend) *Lowlevel {
|
|
db := &Lowlevel{
|
|
Backend: backend,
|
|
folderIdx: newSmallIndex(backend, []byte{KeyTypeFolderIdx}),
|
|
deviceIdx: newSmallIndex(backend, []byte{KeyTypeDeviceIdx}),
|
|
gcMut: sync.NewRWMutex(),
|
|
gcStop: make(chan struct{}),
|
|
}
|
|
db.keyer = newDefaultKeyer(db.folderIdx, db.deviceIdx)
|
|
go db.gcRunner()
|
|
return db
|
|
}
|
|
|
|
func (db *Lowlevel) Close() error {
|
|
close(db.gcStop)
|
|
return db.Backend.Close()
|
|
}
|
|
|
|
// ListFolders returns the list of folders currently in the database
|
|
func (db *Lowlevel) ListFolders() []string {
|
|
return db.folderIdx.Values()
|
|
}
|
|
|
|
// updateRemoteFiles adds a list of fileinfos to the database and updates the
|
|
// global versionlist and metadata.
|
|
func (db *Lowlevel) updateRemoteFiles(folder, device []byte, fs []protocol.FileInfo, meta *metadataTracker) error {
|
|
db.gcMut.RLock()
|
|
defer db.gcMut.RUnlock()
|
|
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
var dk, gk, keyBuf []byte
|
|
devID := protocol.DeviceIDFromBytes(device)
|
|
for _, f := range fs {
|
|
name := []byte(f.Name)
|
|
dk, err = db.keyer.GenerateDeviceFileKey(dk, folder, device, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ef, ok, err := t.getFileTrunc(dk, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ok && unchanged(f, ef) {
|
|
continue
|
|
}
|
|
|
|
if ok {
|
|
meta.removeFile(devID, ef)
|
|
}
|
|
meta.addFile(devID, f)
|
|
|
|
l.Debugf("insert; folder=%q device=%v %v", folder, devID, f)
|
|
if err := t.putFile(dk, f); err != nil {
|
|
return err
|
|
}
|
|
|
|
gk, err = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keyBuf, _, err = t.updateGlobal(gk, keyBuf, folder, device, f, meta)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := t.Checkpoint(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return t.Commit()
|
|
}
|
|
|
|
// updateLocalFiles adds fileinfos to the db, and updates the global versionlist,
|
|
// metadata, sequence and blockmap buckets.
|
|
func (db *Lowlevel) updateLocalFiles(folder []byte, fs []protocol.FileInfo, meta *metadataTracker) error {
|
|
db.gcMut.RLock()
|
|
defer db.gcMut.RUnlock()
|
|
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
var dk, gk, keyBuf []byte
|
|
blockBuf := make([]byte, 4)
|
|
for _, f := range fs {
|
|
name := []byte(f.Name)
|
|
dk, err = db.keyer.GenerateDeviceFileKey(dk, folder, protocol.LocalDeviceID[:], name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ef, ok, err := t.getFileByKey(dk)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ok && unchanged(f, ef) {
|
|
continue
|
|
}
|
|
|
|
if ok {
|
|
if !ef.IsDirectory() && !ef.IsDeleted() && !ef.IsInvalid() {
|
|
for _, block := range ef.Blocks {
|
|
keyBuf, err = db.keyer.GenerateBlockMapKey(keyBuf, folder, block.Hash, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.Delete(keyBuf); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
keyBuf, err = db.keyer.GenerateSequenceKey(keyBuf, folder, ef.SequenceNo())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.Delete(keyBuf); err != nil {
|
|
return err
|
|
}
|
|
l.Debugf("removing sequence; folder=%q sequence=%v %v", folder, ef.SequenceNo(), ef.FileName())
|
|
}
|
|
|
|
f.Sequence = meta.nextLocalSeq()
|
|
|
|
if ok {
|
|
meta.removeFile(protocol.LocalDeviceID, ef)
|
|
}
|
|
meta.addFile(protocol.LocalDeviceID, f)
|
|
|
|
l.Debugf("insert (local); folder=%q %v", folder, f)
|
|
if err := t.putFile(dk, f); err != nil {
|
|
return err
|
|
}
|
|
|
|
gk, err = db.keyer.GenerateGlobalVersionKey(gk, folder, []byte(f.Name))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keyBuf, _, err = t.updateGlobal(gk, keyBuf, folder, protocol.LocalDeviceID[:], f, meta)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
keyBuf, err = db.keyer.GenerateSequenceKey(keyBuf, folder, f.Sequence)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.Put(keyBuf, dk); err != nil {
|
|
return err
|
|
}
|
|
l.Debugf("adding sequence; folder=%q sequence=%v %v", folder, f.Sequence, f.Name)
|
|
|
|
if !f.IsDirectory() && !f.IsDeleted() && !f.IsInvalid() {
|
|
for i, block := range f.Blocks {
|
|
binary.BigEndian.PutUint32(blockBuf, uint32(i))
|
|
keyBuf, err = db.keyer.GenerateBlockMapKey(keyBuf, folder, block.Hash, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.Put(keyBuf, blockBuf); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
if err := t.Checkpoint(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return t.Commit()
|
|
}
|
|
|
|
func (db *Lowlevel) dropFolder(folder []byte) error {
|
|
db.gcMut.RLock()
|
|
defer db.gcMut.RUnlock()
|
|
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
// Remove all items related to the given folder from the device->file bucket
|
|
k0, err := db.keyer.GenerateDeviceFileKey(nil, folder, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(k0.WithoutNameAndDevice()); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Remove all sequences related to the folder
|
|
k1, err := db.keyer.GenerateSequenceKey(nil, folder, 0)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(k1.WithoutSequence()); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Remove all items related to the given folder from the global bucket
|
|
k2, err := db.keyer.GenerateGlobalVersionKey(nil, folder, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(k2.WithoutName()); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Remove all needs related to the folder
|
|
k3, err := db.keyer.GenerateNeedFileKey(nil, folder, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(k3.WithoutName()); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Remove the blockmap of the folder
|
|
k4, err := db.keyer.GenerateBlockMapKey(nil, folder, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(k4.WithoutHashAndName()); err != nil {
|
|
return err
|
|
}
|
|
|
|
return t.Commit()
|
|
}
|
|
|
|
func (db *Lowlevel) dropDeviceFolder(device, folder []byte, meta *metadataTracker) error {
|
|
db.gcMut.RLock()
|
|
defer db.gcMut.RUnlock()
|
|
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
key, err := db.keyer.GenerateDeviceFileKey(nil, folder, device, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
dbi, err := t.NewPrefixIterator(key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var gk, keyBuf []byte
|
|
for dbi.Next() {
|
|
name := db.keyer.NameFromDeviceFileKey(dbi.Key())
|
|
gk, err = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keyBuf, err = t.removeFromGlobal(gk, keyBuf, folder, device, name, meta)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.Delete(dbi.Key()); err != nil {
|
|
return err
|
|
}
|
|
if err := t.Checkpoint(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := dbi.Error(); err != nil {
|
|
return err
|
|
}
|
|
dbi.Release()
|
|
|
|
if bytes.Equal(device, protocol.LocalDeviceID[:]) {
|
|
key, err := db.keyer.GenerateBlockMapKey(nil, folder, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.deleteKeyPrefix(key.WithoutHashAndName()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return t.Commit()
|
|
}
|
|
|
|
func (db *Lowlevel) checkGlobals(folder []byte, meta *metadataTracker) error {
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
key, err := db.keyer.GenerateGlobalVersionKey(nil, folder, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
dbi, err := t.NewPrefixIterator(key.WithoutName())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer dbi.Release()
|
|
|
|
var dk []byte
|
|
for dbi.Next() {
|
|
vl, ok := unmarshalVersionList(dbi.Value())
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
// Check the global version list for consistency. An issue in previous
|
|
// versions of goleveldb could result in reordered writes so that
|
|
// there are global entries pointing to no longer existing files. Here
|
|
// we find those and clear them out.
|
|
|
|
name := db.keyer.NameFromGlobalVersionKey(dbi.Key())
|
|
var newVL VersionList
|
|
for i, version := range vl.Versions {
|
|
dk, err = db.keyer.GenerateDeviceFileKey(dk, folder, version.Device, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err := t.Get(dk)
|
|
if backend.IsNotFound(err) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
newVL.Versions = append(newVL.Versions, version)
|
|
|
|
if i == 0 {
|
|
if fi, ok, err := t.getFileByKey(dk); err != nil {
|
|
return err
|
|
} else if ok {
|
|
meta.addFile(protocol.GlobalDeviceID, fi)
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(newVL.Versions) != len(vl.Versions) {
|
|
if err := t.Put(dbi.Key(), mustMarshal(&newVL)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if err := dbi.Error(); err != nil {
|
|
return err
|
|
}
|
|
|
|
l.Debugf("db check completed for %q", folder)
|
|
return t.Commit()
|
|
}
|
|
|
|
func (db *Lowlevel) getIndexID(device, folder []byte) (protocol.IndexID, error) {
|
|
key, err := db.keyer.GenerateIndexIDKey(nil, device, folder)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
cur, err := db.Get(key)
|
|
if backend.IsNotFound(err) {
|
|
return 0, nil
|
|
} else if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
var id protocol.IndexID
|
|
if err := id.Unmarshal(cur); err != nil {
|
|
return 0, nil
|
|
}
|
|
|
|
return id, nil
|
|
}
|
|
|
|
func (db *Lowlevel) setIndexID(device, folder []byte, id protocol.IndexID) error {
|
|
bs, _ := id.Marshal() // marshalling can't fail
|
|
key, err := db.keyer.GenerateIndexIDKey(nil, device, folder)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return db.Put(key, bs)
|
|
}
|
|
|
|
func (db *Lowlevel) dropMtimes(folder []byte) error {
|
|
key, err := db.keyer.GenerateMtimesKey(nil, folder)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return db.dropPrefix(key)
|
|
}
|
|
|
|
func (db *Lowlevel) dropFolderMeta(folder []byte) error {
|
|
key, err := db.keyer.GenerateFolderMetaKey(nil, folder)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return db.dropPrefix(key)
|
|
}
|
|
|
|
func (db *Lowlevel) dropPrefix(prefix []byte) error {
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer t.close()
|
|
|
|
if err := t.deleteKeyPrefix(prefix); err != nil {
|
|
return err
|
|
}
|
|
return t.Commit()
|
|
}
|
|
|
|
func (db *Lowlevel) gcRunner() {
|
|
t := time.NewTimer(db.timeUntil(blockGCTimeKey, blockGCInterval))
|
|
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)
|
|
}
|
|
db.recordTime(blockGCTimeKey)
|
|
t.Reset(db.timeUntil(blockGCTimeKey, blockGCInterval))
|
|
}
|
|
}
|
|
}
|
|
|
|
// recordTime records the current time under the given key, affecting the
|
|
// next call to timeUntil with the same key.
|
|
func (db *Lowlevel) recordTime(key string) {
|
|
miscDB := NewMiscDataNamespace(db)
|
|
_ = miscDB.PutInt64(key, time.Now().Unix()) // error wilfully ignored
|
|
}
|
|
|
|
// timeUntil returns how long we should wait until the next interval, or
|
|
// zero if it should happen directly.
|
|
func (db *Lowlevel) timeUntil(key string, every time.Duration) time.Duration {
|
|
miscDB := NewMiscDataNamespace(db)
|
|
lastTime, _, _ := miscDB.Int64(key) // error wilfully ignored
|
|
nextTime := time.Unix(lastTime, 0).Add(every)
|
|
sleepTime := time.Until(nextTime)
|
|
if sleepTime < 0 {
|
|
sleepTime = 0
|
|
}
|
|
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).
|
|
//
|
|
// Block GC needs to run when there are no modifications to the FileInfos or
|
|
// block lists.
|
|
|
|
db.gcMut.Lock()
|
|
defer db.gcMut.Unlock()
|
|
|
|
t, err := db.newReadWriteTransaction()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
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.
|
|
|
|
capacity := blockGCBloomCapacity
|
|
if db.gcKeyCount > capacity {
|
|
capacity = db.gcKeyCount
|
|
}
|
|
filter := bloom.NewWithEstimates(uint(capacity), blockGCBloomFalsePositiveRate)
|
|
|
|
// Iterate the FileInfos, unmarshal the blocks hashes and add them to
|
|
// the filter.
|
|
|
|
it, err := db.NewPrefixIterator([]byte{KeyTypeDevice})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
it.Release()
|
|
if err := it.Error(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Iterate over block lists, removing keys with hashes that don't match
|
|
// the filter.
|
|
|
|
it, err = db.NewPrefixIterator([]byte{KeyTypeBlockList})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
matched := 0
|
|
for it.Next() {
|
|
key := blockListKey(it.Key())
|
|
if filter.Test(key.BlocksHash()) {
|
|
matched++
|
|
continue
|
|
}
|
|
if err := t.Delete(key); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
it.Release()
|
|
if err := it.Error(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Remember the number of unique keys we kept until the next pass.
|
|
db.gcKeyCount = matched
|
|
|
|
if err := t.Commit(); err != nil {
|
|
return err
|
|
}
|
|
|
|
return db.Compact()
|
|
}
|
|
|
|
func unmarshalVersionList(data []byte) (VersionList, bool) {
|
|
var vl VersionList
|
|
if err := vl.Unmarshal(data); err != nil {
|
|
l.Debugln("unmarshal error:", err)
|
|
return VersionList{}, false
|
|
}
|
|
if len(vl.Versions) == 0 {
|
|
l.Debugln("empty version list")
|
|
return VersionList{}, false
|
|
}
|
|
return vl, true
|
|
}
|
|
|
|
// unchanged checks if two files are the same and thus don't need to be updated.
|
|
// Local flags or the invalid bit might change without the version
|
|
// being bumped.
|
|
func unchanged(nf, ef FileIntf) bool {
|
|
return ef.FileVersion().Equal(nf.FileVersion()) && ef.IsInvalid() == nf.IsInvalid() && ef.FileLocalFlags() == nf.FileLocalFlags()
|
|
}
|