| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385 |
- // 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"
- "context"
- "encoding/binary"
- "errors"
- "fmt"
- "io"
- "os"
- "regexp"
- "time"
- "github.com/dchest/siphash"
- "github.com/greatroar/blobloom"
- "github.com/syncthing/syncthing/lib/db/backend"
- "github.com/syncthing/syncthing/lib/events"
- "github.com/syncthing/syncthing/lib/fs"
- "github.com/syncthing/syncthing/lib/protocol"
- "github.com/syncthing/syncthing/lib/rand"
- "github.com/syncthing/syncthing/lib/sha256"
- "github.com/syncthing/syncthing/lib/svcutil"
- "github.com/syncthing/syncthing/lib/sync"
- "github.com/syncthing/syncthing/lib/util"
- "github.com/thejerf/suture/v4"
- )
- const (
- // 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%
- indirectGCBloomMaxBytes = 32 << 20 // Use at most 32MiB memory, which covers our desired FP rate at 27 M items
- indirectGCDefaultInterval = 13 * time.Hour
- indirectGCTimeKey = "lastIndirectGCTime"
- // Use indirection for the block list when it exceeds this many entries
- blocksIndirectionCutoff = 3
- // Use indirection for the version vector when it exceeds this many entries
- versionIndirectionCutoff = 10
- recheckDefaultInterval = 30 * 24 * time.Hour
- needsRepairSuffix = ".needsrepair"
- )
- // 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 {
- *suture.Supervisor
- backend.Backend
- folderIdx *smallIndex
- deviceIdx *smallIndex
- keyer keyer
- gcMut sync.RWMutex
- gcKeyCount int
- indirectGCInterval time.Duration
- recheckInterval time.Duration
- oneFileSetCreated chan struct{}
- evLogger events.Logger
- blockFilter *bloomFilter
- versionFilter *bloomFilter
- }
- func NewLowlevel(backend backend.Backend, evLogger events.Logger, opts ...Option) (*Lowlevel, error) {
- // Only log restarts in debug mode.
- spec := svcutil.SpecWithDebugLogger(l)
- db := &Lowlevel{
- Supervisor: suture.New("db.Lowlevel", spec),
- Backend: backend,
- folderIdx: newSmallIndex(backend, []byte{KeyTypeFolderIdx}),
- deviceIdx: newSmallIndex(backend, []byte{KeyTypeDeviceIdx}),
- gcMut: sync.NewRWMutex(),
- indirectGCInterval: indirectGCDefaultInterval,
- recheckInterval: recheckDefaultInterval,
- oneFileSetCreated: make(chan struct{}),
- evLogger: evLogger,
- }
- for _, opt := range opts {
- opt(db)
- }
- db.keyer = newDefaultKeyer(db.folderIdx, db.deviceIdx)
- db.Add(svcutil.AsService(db.gcRunner, "db.Lowlevel/gcRunner"))
- if path := db.needsRepairPath(); path != "" {
- if _, err := os.Lstat(path); err == nil {
- l.Infoln("Database was marked for repair - this may take a while")
- if err := db.checkRepair(); err != nil {
- db.handleFailure(err)
- return nil, err
- }
- os.Remove(path)
- }
- }
- return db, nil
- }
- type Option func(*Lowlevel)
- // WithRecheckInterval sets the time interval in between metadata recalculations
- // and consistency checks.
- func WithRecheckInterval(dur time.Duration) Option {
- return func(db *Lowlevel) {
- if dur > 0 {
- db.recheckInterval = dur
- }
- }
- }
- // WithIndirectGCInterval sets the time interval in between GC runs.
- func WithIndirectGCInterval(dur time.Duration) Option {
- return func(db *Lowlevel) {
- if dur > 0 {
- db.indirectGCInterval = dur
- }
- }
- }
- // 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(meta.CommitHook(folder))
- if err != nil {
- return err
- }
- defer t.close()
- var dk, gk, keyBuf []byte
- devID, err := protocol.DeviceIDFromBytes(device)
- if err != nil {
- return err
- }
- 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) {
- l.Debugf("not inserting unchanged (remote); folder=%q device=%v %v", folder, devID, f)
- continue
- }
- if ok {
- meta.removeFile(devID, ef)
- }
- meta.addFile(devID, f)
- l.Debugf("insert (remote); 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(meta.CommitHook(folder))
- 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) {
- l.Debugf("not inserting unchanged (local); folder=%q %v", folder, f)
- continue
- }
- blocksHashSame := ok && bytes.Equal(ef.BlocksHash, f.BlocksHash)
- if ok {
- if len(ef.Blocks) != 0 && !ef.IsInvalid() && ef.Size > 0 {
- 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
- }
- }
- if !blocksHashSame {
- keyBuf, err := db.keyer.GenerateBlockListMapKey(keyBuf, folder, ef.BlocksHash, 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 len(f.Blocks) != 0 && !f.IsInvalid() && f.Size > 0 {
- 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 !blocksHashSame {
- keyBuf, err := db.keyer.GenerateBlockListMapKey(keyBuf, folder, f.BlocksHash, name)
- if err != nil {
- return err
- }
- if err = t.Put(keyBuf, nil); 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(k0, 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(k1, 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(k2, 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(k3, folder, nil, nil)
- if err != nil {
- return err
- }
- if err := t.deleteKeyPrefix(k4.WithoutHashAndName()); err != nil {
- return err
- }
- k5, err := db.keyer.GenerateBlockListMapKey(k4, folder, nil, nil)
- if err != nil {
- return err
- }
- if err := t.deleteKeyPrefix(k5.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(meta.CommitHook(folder))
- 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
- }
- defer dbi.Release()
- 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
- }
- }
- dbi.Release()
- if err := dbi.Error(); err != nil {
- return err
- }
- 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
- }
- key2, err := db.keyer.GenerateBlockListMapKey(key, folder, nil, nil)
- if err != nil {
- return err
- }
- if err := t.deleteKeyPrefix(key2.WithoutHashAndName()); err != nil {
- return err
- }
- }
- return t.Commit()
- }
- func (db *Lowlevel) checkGlobals(folderStr string) (int, error) {
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return 0, err
- }
- defer t.close()
- folder := []byte(folderStr)
- key, err := db.keyer.GenerateGlobalVersionKey(nil, folder, nil)
- if err != nil {
- return 0, err
- }
- dbi, err := t.NewPrefixIterator(key.WithoutName())
- if err != nil {
- return 0, err
- }
- defer dbi.Release()
- fixed := 0
- var dk []byte
- ro := t.readOnlyTransaction
- for dbi.Next() {
- var vl VersionList
- if err := vl.Unmarshal(dbi.Value()); err != nil || vl.Empty() {
- if err := t.Delete(dbi.Key()); err != nil && !backend.IsNotFound(err) {
- return 0, err
- }
- 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())
- newVL := &VersionList{}
- var changed, changedHere bool
- for _, fv := range vl.RawVersions {
- changedHere, err = checkGlobalsFilterDevices(dk, folder, name, fv.Devices, newVL, ro)
- if err != nil {
- return 0, err
- }
- changed = changed || changedHere
- changedHere, err = checkGlobalsFilterDevices(dk, folder, name, fv.InvalidDevices, newVL, ro)
- if err != nil {
- return 0, err
- }
- changed = changed || changedHere
- }
- if newVL.Empty() {
- if err := t.Delete(dbi.Key()); err != nil && !backend.IsNotFound(err) {
- return 0, err
- }
- fixed++
- } else if changed {
- if err := t.Put(dbi.Key(), mustMarshal(newVL)); err != nil {
- return 0, err
- }
- fixed++
- }
- }
- dbi.Release()
- if err := dbi.Error(); err != nil {
- return 0, err
- }
- l.Debugf("global db check completed for %v", folder)
- return fixed, t.Commit()
- }
- func checkGlobalsFilterDevices(dk, folder, name []byte, devices [][]byte, vl *VersionList, t readOnlyTransaction) (bool, error) {
- var changed bool
- var err error
- for _, device := range devices {
- dk, err = t.keyer.GenerateDeviceFileKey(dk, folder, device, name)
- if err != nil {
- return false, err
- }
- f, ok, err := t.getFileTrunc(dk, false)
- if err != nil {
- return false, err
- }
- if !ok {
- changed = true
- continue
- }
- _, _, _, _, _, _, err = vl.update(folder, device, f, t)
- if err != nil {
- return false, err
- }
- }
- return changed, nil
- }
- 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) dropFolderIndexIDs(folder []byte) error {
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return err
- }
- defer t.close()
- if err := t.deleteKeyPrefixMatching([]byte{KeyTypeIndexID}, func(key []byte) bool {
- keyFolder, ok := t.keyer.FolderFromIndexIDKey(key)
- if !ok {
- l.Debugf("Deleting IndexID with missing FolderIdx: %v", key)
- return true
- }
- return bytes.Equal(keyFolder, folder)
- }); err != nil {
- return err
- }
- return t.Commit()
- }
- func (db *Lowlevel) dropIndexIDs() error {
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return err
- }
- defer t.close()
- if err := t.deleteKeyPrefix([]byte{KeyTypeIndexID}); err != nil {
- return err
- }
- return t.Commit()
- }
- 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(ctx context.Context) error {
- // 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, db.indirectGCInterval)
- if next < time.Minute {
- next = time.Minute
- }
- t := time.NewTimer(next)
- defer t.Stop()
- for {
- select {
- case <-ctx.Done():
- return ctx.Err()
- case <-t.C:
- if err := db.gcIndirect(ctx); err != nil {
- l.Warnln("Database indirection GC failed:", err)
- }
- db.recordTime(indirectGCTimeKey)
- t.Reset(db.timeUntil(indirectGCTimeKey, db.indirectGCInterval))
- }
- }
- }
- // 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) gcIndirect(ctx context.Context) (err 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).
- //
- // Indirection GC needs to run when there are no modifications to the
- // FileInfos or indirected items.
- l.Debugln("Starting database GC")
- // Create a new set of bloom filters, while holding the gcMut which
- // guarantees that no other modifications are happening concurrently.
- db.gcMut.Lock()
- capacity := indirectGCBloomCapacity
- if db.gcKeyCount > capacity {
- capacity = db.gcKeyCount
- }
- db.blockFilter = newBloomFilter(capacity)
- db.versionFilter = newBloomFilter(capacity)
- db.gcMut.Unlock()
- defer func() {
- // Forget the bloom filters on the way out.
- db.gcMut.Lock()
- db.blockFilter = nil
- db.versionFilter = nil
- db.gcMut.Unlock()
- }()
- var discardedBlocks, matchedBlocks, discardedVersions, matchedVersions int
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return err
- }
- defer t.Release()
- // 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.
- // Iterate the FileInfos, unmarshal the block and version hashes and
- // add them to the filter.
- // This happens concurrently with normal database modifications, though
- // those modifications will now also add their blocks and versions to
- // the bloom filters.
- it, err := t.NewPrefixIterator([]byte{KeyTypeDevice})
- if err != nil {
- return err
- }
- defer it.Release()
- for it.Next() {
- select {
- case <-ctx.Done():
- return ctx.Err()
- default:
- }
- var hashes IndirectionHashesOnly
- if err := hashes.Unmarshal(it.Value()); err != nil {
- return err
- }
- db.recordIndirectionHashes(hashes)
- }
- it.Release()
- if err := it.Error(); err != nil {
- return err
- }
- // For the next phase we grab the GC lock again and hold it for the rest
- // of the method call. Now there can't be any further modifications to
- // the database or the bloom filters.
- db.gcMut.Lock()
- defer db.gcMut.Unlock()
- // Only print something if the process takes more than "a moment".
- logWait := make(chan struct{})
- logTimer := time.AfterFunc(10*time.Second, func() {
- l.Infoln("Database GC in progress - many Syncthing operations will be unresponsive until it's finished")
- close(logWait)
- })
- defer func() {
- if logTimer.Stop() {
- return
- }
- <-logWait // Make sure messages are sent in order.
- l.Infof("Database GC complete (discarded/remaining: %v/%v blocks, %v/%v versions)",
- discardedBlocks, matchedBlocks, discardedVersions, matchedVersions)
- }()
- // Iterate over block lists, removing keys with hashes that don't match
- // the filter.
- it, err = t.NewPrefixIterator([]byte{KeyTypeBlockList})
- if err != nil {
- return err
- }
- defer it.Release()
- for it.Next() {
- select {
- case <-ctx.Done():
- return ctx.Err()
- default:
- }
- key := blockListKey(it.Key())
- if db.blockFilter.has(key.Hash()) {
- matchedBlocks++
- continue
- }
- if err := t.Delete(key); err != nil {
- return err
- }
- discardedBlocks++
- }
- it.Release()
- if err := it.Error(); err != nil {
- return err
- }
- // Iterate over version lists, removing keys with hashes that don't match
- // the filter.
- it, err = db.NewPrefixIterator([]byte{KeyTypeVersion})
- if err != nil {
- return err
- }
- for it.Next() {
- select {
- case <-ctx.Done():
- return ctx.Err()
- default:
- }
- key := versionKey(it.Key())
- if db.versionFilter.has(key.Hash()) {
- matchedVersions++
- continue
- }
- if err := t.Delete(key); err != nil {
- return err
- }
- discardedVersions++
- }
- it.Release()
- if err := it.Error(); err != nil {
- return err
- }
- // Remember the number of unique keys we kept until the next pass.
- db.gcKeyCount = matchedBlocks
- if matchedVersions > matchedBlocks {
- db.gcKeyCount = matchedVersions
- }
- if err := t.Commit(); err != nil {
- return err
- }
- l.Debugf("Finished GC (discarded/remaining: %v/%v blocks, %v/%v versions)", discardedBlocks, matchedBlocks, discardedVersions, matchedVersions)
- return nil
- }
- func (db *Lowlevel) recordIndirectionHashesForFile(f *protocol.FileInfo) {
- db.recordIndirectionHashes(IndirectionHashesOnly{BlocksHash: f.BlocksHash, VersionHash: f.VersionHash})
- }
- func (db *Lowlevel) recordIndirectionHashes(hs IndirectionHashesOnly) {
- // must be called with gcMut held (at least read-held)
- if db.blockFilter != nil && len(hs.BlocksHash) > 0 {
- db.blockFilter.add(hs.BlocksHash)
- }
- if db.versionFilter != nil && len(hs.VersionHash) > 0 {
- db.versionFilter.add(hs.VersionHash)
- }
- }
- func newBloomFilter(capacity int) *bloomFilter {
- var buf [16]byte
- io.ReadFull(rand.Reader, buf[:])
- return &bloomFilter{
- f: blobloom.NewSyncOptimized(blobloom.Config{
- Capacity: uint64(capacity),
- FPRate: indirectGCBloomFalsePositiveRate,
- MaxBits: 8 * indirectGCBloomMaxBytes,
- }),
- k0: binary.LittleEndian.Uint64(buf[:8]),
- k1: binary.LittleEndian.Uint64(buf[8:]),
- }
- }
- type bloomFilter struct {
- f *blobloom.SyncFilter
- k0, k1 uint64 // Random key for SipHash.
- }
- func (b *bloomFilter) add(id []byte) { b.f.Add(b.hash(id)) }
- func (b *bloomFilter) has(id []byte) bool { return b.f.Has(b.hash(id)) }
- // Hash function for the bloomfilter: SipHash of the SHA-256.
- //
- // The randomization in SipHash means we get different collisions across
- // runs and colliding keys are not kept indefinitely.
- func (b *bloomFilter) hash(id []byte) uint64 {
- if len(id) != sha256.Size {
- panic("bug: bloomFilter.hash passed something not a SHA256 hash")
- }
- return siphash.Hash(b.k0, b.k1, id)
- }
- // checkRepair checks folder metadata and sequences for miscellaneous errors.
- func (db *Lowlevel) checkRepair() error {
- db.gcMut.RLock()
- defer db.gcMut.RUnlock()
- for _, folder := range db.ListFolders() {
- if _, err := db.getMetaAndCheckGCLocked(folder); err != nil {
- return err
- }
- }
- return nil
- }
- func (db *Lowlevel) getMetaAndCheck(folder string) (*metadataTracker, error) {
- db.gcMut.RLock()
- defer db.gcMut.RUnlock()
- return db.getMetaAndCheckGCLocked(folder)
- }
- func (db *Lowlevel) getMetaAndCheckGCLocked(folder string) (*metadataTracker, error) {
- fixed, err := db.checkLocalNeed([]byte(folder))
- if err != nil {
- return nil, fmt.Errorf("checking local need: %w", err)
- }
- if fixed != 0 {
- l.Infof("Repaired %d local need entries for folder %v in database", fixed, folder)
- }
- fixed, err = db.checkGlobals(folder)
- if err != nil {
- return nil, fmt.Errorf("checking globals: %w", err)
- }
- if fixed != 0 {
- l.Infof("Repaired %d global entries for folder %v in database", fixed, folder)
- }
- oldMeta := newMetadataTracker(db.keyer, db.evLogger)
- _ = oldMeta.fromDB(db, []byte(folder)) // Ignore error, it leads to index id reset too
- meta, err := db.recalcMeta(folder)
- if err != nil {
- return nil, fmt.Errorf("recalculating metadata: %w", err)
- }
- fixed, err = db.repairSequenceGCLocked(folder, meta)
- if err != nil {
- return nil, fmt.Errorf("repairing sequences: %w", err)
- }
- if fixed != 0 {
- l.Infof("Repaired %d sequence entries for folder %v in database", fixed, folder)
- meta, err = db.recalcMeta(folder)
- if err != nil {
- return nil, fmt.Errorf("recalculating metadata: %w", err)
- }
- }
- if err := db.checkSequencesUnchanged(folder, oldMeta, meta); err != nil {
- return nil, fmt.Errorf("checking for changed sequences: %w", err)
- }
- return meta, nil
- }
- func (db *Lowlevel) loadMetadataTracker(folder string) (*metadataTracker, error) {
- meta := newMetadataTracker(db.keyer, db.evLogger)
- if err := meta.fromDB(db, []byte(folder)); err != nil {
- if err == errMetaInconsistent {
- l.Infof("Stored folder metadata for %q is inconsistent; recalculating", folder)
- } else {
- l.Infof("No stored folder metadata for %q; recalculating", folder)
- }
- return db.getMetaAndCheck(folder)
- }
- curSeq := meta.Sequence(protocol.LocalDeviceID)
- if metaOK, err := db.verifyLocalSequence(curSeq, folder); err != nil {
- return nil, fmt.Errorf("verifying sequences: %w", err)
- } else if !metaOK {
- l.Infof("Stored folder metadata for %q is out of date after crash; recalculating", folder)
- return db.getMetaAndCheck(folder)
- }
- if age := time.Since(meta.Created()); age > db.recheckInterval {
- l.Infof("Stored folder metadata for %q is %v old; recalculating", folder, util.NiceDurationString(age))
- return db.getMetaAndCheck(folder)
- }
- return meta, nil
- }
- func (db *Lowlevel) recalcMeta(folderStr string) (*metadataTracker, error) {
- folder := []byte(folderStr)
- meta := newMetadataTracker(db.keyer, db.evLogger)
- t, err := db.newReadWriteTransaction(meta.CommitHook(folder))
- if err != nil {
- return nil, err
- }
- defer t.close()
- var deviceID protocol.DeviceID
- err = t.withAllFolderTruncated(folder, func(device []byte, f FileInfoTruncated) bool {
- copy(deviceID[:], device)
- meta.addFile(deviceID, f)
- return true
- })
- if err != nil {
- return nil, err
- }
- err = t.withGlobal(folder, nil, true, func(f protocol.FileIntf) bool {
- meta.addFile(protocol.GlobalDeviceID, f)
- return true
- })
- if err != nil {
- return nil, err
- }
- meta.emptyNeeded(protocol.LocalDeviceID)
- err = t.withNeed(folder, protocol.LocalDeviceID[:], true, func(f protocol.FileIntf) bool {
- meta.addNeeded(protocol.LocalDeviceID, f)
- return true
- })
- if err != nil {
- return nil, err
- }
- for _, device := range meta.devices() {
- meta.emptyNeeded(device)
- err = t.withNeed(folder, device[:], true, func(f protocol.FileIntf) bool {
- meta.addNeeded(device, f)
- return true
- })
- if err != nil {
- return nil, err
- }
- }
- meta.SetCreated()
- 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 (db *Lowlevel) verifyLocalSequence(curSeq int64, folder string) (bool, error) {
- // 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 {
- return false, err
- }
- ok := true
- if err := t.withHaveSequence([]byte(folder), curSeq+1, func(fi protocol.FileIntf) bool {
- ok = false // we got something, which we should not have
- return false
- }); err != nil {
- return false, err
- }
- t.close()
- return ok, nil
- }
- // repairSequenceGCLocked makes sure the sequence numbers in the sequence keys
- // match those in the corresponding file entries. It returns the amount of fixed
- // entries.
- func (db *Lowlevel) repairSequenceGCLocked(folderStr string, meta *metadataTracker) (int, error) {
- t, err := db.newReadWriteTransaction(meta.CommitHook([]byte(folderStr)))
- if err != nil {
- return 0, err
- }
- defer t.close()
- fixed := 0
- folder := []byte(folderStr)
- // First check that every file entry has a matching sequence entry
- // (this was previously db schema upgrade to 9).
- dk, err := t.keyer.GenerateDeviceFileKey(nil, folder, protocol.LocalDeviceID[:], nil)
- if err != nil {
- return 0, err
- }
- it, err := t.NewPrefixIterator(dk.WithoutName())
- if err != nil {
- return 0, err
- }
- defer it.Release()
- var sk sequenceKey
- for it.Next() {
- intf, err := t.unmarshalTrunc(it.Value(), false)
- if err != nil {
- // Delete local items with invalid indirected blocks/versions.
- // They will be rescanned.
- var ierr *blocksIndirectionError
- if ok := errors.As(err, &ierr); ok && backend.IsNotFound(err) {
- intf, err = t.unmarshalTrunc(it.Value(), true)
- if err != nil {
- return 0, err
- }
- name := []byte(intf.FileName())
- gk, err := t.keyer.GenerateGlobalVersionKey(nil, folder, name)
- if err != nil {
- return 0, err
- }
- _, err = t.removeFromGlobal(gk, nil, folder, protocol.LocalDeviceID[:], name, nil)
- if err != nil {
- return 0, err
- }
- sk, err = db.keyer.GenerateSequenceKey(sk, folder, intf.SequenceNo())
- if err != nil {
- return 0, err
- }
- if err := t.Delete(sk); err != nil {
- return 0, err
- }
- if err := t.Delete(it.Key()); err != nil {
- return 0, err
- }
- }
- return 0, err
- }
- fi := intf.(protocol.FileInfo)
- if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
- return 0, err
- }
- switch dk, err = t.Get(sk); {
- case err != nil:
- if !backend.IsNotFound(err) {
- return 0, err
- }
- fallthrough
- case !bytes.Equal(it.Key(), dk):
- fixed++
- fi.Sequence = meta.nextLocalSeq()
- if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
- return 0, err
- }
- if err := t.Put(sk, it.Key()); err != nil {
- return 0, err
- }
- if err := t.putFile(it.Key(), fi); err != nil {
- return 0, err
- }
- }
- if err := t.Checkpoint(); err != nil {
- return 0, err
- }
- }
- if err := it.Error(); err != nil {
- return 0, err
- }
- it.Release()
- // Secondly check there's no sequence entries pointing at incorrect things.
- sk, err = t.keyer.GenerateSequenceKey(sk, folder, 0)
- if err != nil {
- return 0, err
- }
- it, err = t.NewPrefixIterator(sk.WithoutSequence())
- if err != nil {
- return 0, err
- }
- defer it.Release()
- for it.Next() {
- // Check that the sequence from the key matches the
- // sequence in the file.
- fi, ok, err := t.getFileTrunc(it.Value(), true)
- if err != nil {
- return 0, err
- }
- if ok {
- if seq := t.keyer.SequenceFromSequenceKey(it.Key()); seq == fi.SequenceNo() {
- continue
- }
- }
- // Either the file is missing or has a different sequence number
- fixed++
- if err := t.Delete(it.Key()); err != nil {
- return 0, err
- }
- }
- if err := it.Error(); err != nil {
- return 0, err
- }
- it.Release()
- return fixed, t.Commit()
- }
- // Does not take care of metadata - if anything is repaired, the need count
- // needs to be recalculated.
- func (db *Lowlevel) checkLocalNeed(folder []byte) (int, error) {
- repaired := 0
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return 0, err
- }
- defer t.close()
- key, err := t.keyer.GenerateNeedFileKey(nil, folder, nil)
- if err != nil {
- return 0, err
- }
- dbi, err := t.NewPrefixIterator(key.WithoutName())
- if err != nil {
- return 0, err
- }
- defer dbi.Release()
- var needName string
- var needDone bool
- next := func() {
- needDone = !dbi.Next()
- if !needDone {
- needName = string(t.keyer.NameFromGlobalVersionKey(dbi.Key()))
- }
- }
- next()
- t.withNeedIteratingGlobal(folder, protocol.LocalDeviceID[:], true, func(fi protocol.FileIntf) bool {
- f := fi.(FileInfoTruncated)
- for !needDone && needName < f.Name {
- repaired++
- if err = t.Delete(dbi.Key()); err != nil && !backend.IsNotFound(err) {
- return false
- }
- l.Debugln("check local need: removing", needName)
- next()
- }
- if needName == f.Name {
- next()
- } else {
- repaired++
- key, err = t.keyer.GenerateNeedFileKey(key, folder, []byte(f.Name))
- if err != nil {
- return false
- }
- if err = t.Put(key, nil); err != nil {
- return false
- }
- l.Debugln("check local need: adding", f.Name)
- }
- return true
- })
- if err != nil {
- return 0, err
- }
- for !needDone {
- repaired++
- if err := t.Delete(dbi.Key()); err != nil && !backend.IsNotFound(err) {
- return 0, err
- }
- l.Debugln("check local need: removing", needName)
- next()
- }
- if err := dbi.Error(); err != nil {
- return 0, err
- }
- dbi.Release()
- if err = t.Commit(); err != nil {
- return 0, err
- }
- return repaired, nil
- }
- // checkSequencesUnchanged resets delta indexes for any device where the
- // sequence changed.
- func (db *Lowlevel) checkSequencesUnchanged(folder string, oldMeta, meta *metadataTracker) error {
- t, err := db.newReadWriteTransaction()
- if err != nil {
- return err
- }
- defer t.close()
- var key []byte
- deleteIndexID := func(devID protocol.DeviceID) error {
- key, err = db.keyer.GenerateIndexIDKey(key, devID[:], []byte(folder))
- if err != nil {
- return err
- }
- return t.Delete(key)
- }
- if oldMeta.Sequence(protocol.LocalDeviceID) != meta.Sequence(protocol.LocalDeviceID) {
- if err := deleteIndexID(protocol.LocalDeviceID); err != nil {
- return err
- }
- l.Infof("Local sequence for folder %v changed while repairing - dropping delta indexes", folder)
- }
- oldDevices := oldMeta.devices()
- oldSequences := make(map[protocol.DeviceID]int64, len(oldDevices))
- for _, devID := range oldDevices {
- oldSequences[devID] = oldMeta.Sequence(devID)
- }
- for _, devID := range meta.devices() {
- oldSeq := oldSequences[devID]
- delete(oldSequences, devID)
- // A lower sequence number just means we will receive some indexes again.
- if oldSeq >= meta.Sequence(devID) {
- if oldSeq > meta.Sequence(devID) {
- db.evLogger.Log(events.Failure, "lower remote sequence after recalculating metadata")
- }
- continue
- }
- db.evLogger.Log(events.Failure, "higher remote sequence after recalculating metadata")
- if err := deleteIndexID(devID); err != nil {
- return err
- }
- l.Infof("Sequence of device %v for folder %v changed while repairing - dropping delta indexes", devID.Short(), folder)
- }
- for devID := range oldSequences {
- if err := deleteIndexID(devID); err != nil {
- return err
- }
- l.Debugf("Removed indexID of device %v for folder %v which isn't present anymore", devID.Short(), folder)
- }
- return t.Commit()
- }
- func (db *Lowlevel) needsRepairPath() string {
- path := db.Location()
- if path == "" {
- return ""
- }
- if path[len(path)-1] == fs.PathSeparator {
- path = path[:len(path)-1]
- }
- return path + needsRepairSuffix
- }
- func (db *Lowlevel) checkErrorForRepair(err error) {
- if errors.Is(err, errEntryFromGlobalMissing) || errors.Is(err, errEmptyGlobal) {
- // Inconsistency error, mark db for repair on next start.
- if path := db.needsRepairPath(); path != "" {
- if fd, err := os.Create(path); err == nil {
- fd.Close()
- }
- }
- }
- }
- // 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 protocol.FileIntf) bool {
- return ef.FileVersion().Equal(nf.FileVersion()) && ef.IsInvalid() == nf.IsInvalid() && ef.FileLocalFlags() == nf.FileLocalFlags()
- }
- func (db *Lowlevel) handleFailure(err error) {
- db.checkErrorForRepair(err)
- if shouldReportFailure(err) {
- db.evLogger.Log(events.Failure, err.Error())
- }
- }
- var ldbPathRe = regexp.MustCompile(`(open|write|read) .+[\\/].+[\\/]index[^\\/]+[\\/][^\\/]+: `)
- func shouldReportFailure(err error) bool {
- return !ldbPathRe.MatchString(err.Error())
- }
|