chore: switch database engine to sqlite (fixes #9954) (#9965)

Switch the database from LevelDB to SQLite, for greater stability and simpler code. Co-authored-by: Tommy van der Vorst <tommy@pixelspark.nl> Co-authored-by: bt90 <btom1990@googlemail.com>
2025-03-29 13:50:08 +01:00
parent b1c8f88a44
commit 025905fcdf
146 changed files with 8315 additions and 11984 deletions
@@ -8,12 +8,14 @@ package model

 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"

+	"github.com/syncthing/syncthing/internal/db"
+	"github.com/syncthing/syncthing/internal/itererr"
 	"github.com/syncthing/syncthing/lib/config"
-	"github.com/syncthing/syncthing/lib/db"
 	"github.com/syncthing/syncthing/lib/events"
 	"github.com/syncthing/syncthing/lib/protocol"
 	"github.com/syncthing/syncthing/lib/svcutil"
@@ -45,13 +47,19 @@ type indexHandler struct {

 	cond   *sync.Cond
 	paused bool
-	fset   *db.FileSet
+	sdb    db.DB
 	runner service
 }

-func newIndexHandler(conn protocol.Connection, downloads *deviceDownloadState, folder config.FolderConfiguration, fset *db.FileSet, runner service, startInfo *clusterConfigDeviceInfo, evLogger events.Logger) *indexHandler {
-	myIndexID := fset.IndexID(protocol.LocalDeviceID)
-	mySequence := fset.Sequence(protocol.LocalDeviceID)
+func newIndexHandler(conn protocol.Connection, downloads *deviceDownloadState, folder config.FolderConfiguration, sdb db.DB, runner service, startInfo *clusterConfigDeviceInfo, evLogger events.Logger) (*indexHandler, error) {
+	myIndexID, err := sdb.GetIndexID(folder.ID, protocol.LocalDeviceID)
+	if err != nil {
+		return nil, err
+	}
+	mySequence, err := sdb.GetDeviceSequence(folder.ID, protocol.LocalDeviceID)
+	if err != nil {
+		return nil, err
+	}
 	var startSequence int64

 	// This is the other side's description of what it knows
@@ -91,14 +99,14 @@ func newIndexHandler(conn protocol.Connection, downloads *deviceDownloadState, f
 	// otherwise we drop our old index data and expect to get a
 	// completely new set.

-	theirIndexID := fset.IndexID(conn.DeviceID())
+	theirIndexID, _ := sdb.GetIndexID(folder.ID, conn.DeviceID())
 	if startInfo.remote.IndexID == 0 {
 		// They're not announcing an index ID. This means they
 		// do not support delta indexes and we should clear any
 		// information we have from them before accepting their
 		// index, which will presumably be a full index.
 		l.Debugf("Device %v folder %s does not announce an index ID", conn.DeviceID().Short(), folder.Description())
-		fset.Drop(conn.DeviceID())
+		sdb.DropAllFiles(folder.ID, conn.DeviceID())
 	} else if startInfo.remote.IndexID != theirIndexID {
 		// The index ID we have on file is not what they're
 		// announcing. They must have reset their database and
@@ -106,8 +114,8 @@ func newIndexHandler(conn protocol.Connection, downloads *deviceDownloadState, f
 		// information we have and remember this new index ID
 		// instead.
 		l.Infof("Device %v folder %s has a new index ID (%v)", conn.DeviceID().Short(), folder.Description(), startInfo.remote.IndexID)
-		fset.Drop(conn.DeviceID())
-		fset.SetIndexID(conn.DeviceID(), startInfo.remote.IndexID)
+		sdb.DropAllFiles(folder.ID, conn.DeviceID())
+		sdb.SetIndexID(folder.ID, conn.DeviceID(), startInfo.remote.IndexID)
 	}

 	return &indexHandler{
@@ -119,27 +127,27 @@ func newIndexHandler(conn protocol.Connection, downloads *deviceDownloadState, f
 		sentPrevSequence:         startSequence,
 		evLogger:                 evLogger,

-		fset:   fset,
+		sdb:    sdb,
 		runner: runner,
 		cond:   sync.NewCond(new(sync.Mutex)),
-	}
+	}, nil
 }

-// waitForFileset waits for the handler to resume and fetches the current fileset.
-func (s *indexHandler) waitForFileset(ctx context.Context) (*db.FileSet, error) {
+// waitWhilePaused waits for the handler to resume
+func (s *indexHandler) waitWhilePaused(ctx context.Context) error {
 	s.cond.L.Lock()
 	defer s.cond.L.Unlock()

 	for s.paused {
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return ctx.Err()
 		default:
 			s.cond.Wait()
 		}
 	}

-	return s.fset, nil
+	return nil
 }

 func (s *indexHandler) Serve(ctx context.Context) (err error) {
@@ -162,11 +170,10 @@ func (s *indexHandler) Serve(ctx context.Context) (err error) {
 	}()

 	// We need to send one index, regardless of whether there is something to send or not
-	fset, err := s.waitForFileset(ctx)
-	if err != nil {
+	if err := s.waitWhilePaused(ctx); err != nil {
 		return err
 	}
-	err = s.sendIndexTo(ctx, fset)
+	err = s.sendIndexTo(ctx)

 	// Subscribe to LocalIndexUpdated (we have new information to send) and
 	// DeviceDisconnected (it might be us who disconnected, so we should
@@ -179,8 +186,7 @@ func (s *indexHandler) Serve(ctx context.Context) (err error) {
 	defer ticker.Stop()

 	for err == nil {
-		fset, err = s.waitForFileset(ctx)
-		if err != nil {
+		if err := s.waitWhilePaused(ctx); err != nil {
 			return err
 		}

@@ -188,7 +194,11 @@ func (s *indexHandler) Serve(ctx context.Context) (err error) {
 		// currently in the database, wait for the local index to update. The
 		// local index may update for other folders than the one we are
 		// sending for.
-		if fset.Sequence(protocol.LocalDeviceID) <= s.localPrevSequence {
+		seq, err := s.sdb.GetDeviceSequence(s.folder, protocol.LocalDeviceID)
+		if err != nil {
+			return err
+		}
+		if seq <= s.localPrevSequence {
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
@@ -198,7 +208,7 @@ func (s *indexHandler) Serve(ctx context.Context) (err error) {
 			continue
 		}

-		err = s.sendIndexTo(ctx, fset)
+		err = s.sendIndexTo(ctx)

 		// Wait a short amount of time before entering the next loop. If there
 		// are continuous changes happening to the local index, this gives us
@@ -215,10 +225,9 @@ func (s *indexHandler) Serve(ctx context.Context) (err error) {

 // resume might be called because the folder was actually resumed, or just
 // because the folder config changed (and thus the runner and potentially fset).
-func (s *indexHandler) resume(fset *db.FileSet, runner service) {
+func (s *indexHandler) resume(runner service) {
 	s.cond.L.Lock()
 	s.paused = false
-	s.fset = fset
 	s.runner = runner
 	s.cond.Broadcast()
 	s.cond.L.Unlock()
@@ -230,7 +239,6 @@ func (s *indexHandler) pause() {
 		s.evLogger.Log(events.Failure, "index handler got paused while already paused")
 	}
 	s.paused = true
-	s.fset = nil
 	s.runner = nil
 	s.cond.Broadcast()
 	s.cond.L.Unlock()
@@ -238,9 +246,9 @@ func (s *indexHandler) pause() {

 // sendIndexTo sends file infos with a sequence number higher than prevSequence and
 // returns the highest sent sequence number.
-func (s *indexHandler) sendIndexTo(ctx context.Context, fset *db.FileSet) error {
+func (s *indexHandler) sendIndexTo(ctx context.Context) error {
 	initial := s.localPrevSequence == 0
-	batch := db.NewFileInfoBatch(nil)
+	batch := NewFileInfoBatch(nil)
 	var batchError error
 	batch.SetFlushFunc(func(fs []protocol.FileInfo) error {
 		select {
@@ -284,21 +292,26 @@ func (s *indexHandler) sendIndexTo(ctx context.Context, fset *db.FileSet) error
 		return nil
 	})

-	var err error
 	var f protocol.FileInfo
-	snap, err := fset.Snapshot()
-	if err != nil {
-		return svcutil.AsFatalErr(err, svcutil.ExitError)
-	}
-	defer snap.Release()
 	previousWasDelete := false
-	snap.WithHaveSequence(s.localPrevSequence+1, func(fi protocol.FileInfo) bool {
+
+	t0 := time.Now()
+	for fi, err := range itererr.Zip(s.sdb.AllLocalFilesBySequence(s.folder, protocol.LocalDeviceID, s.localPrevSequence+1, 5000)) {
+		if err != nil {
+			return err
+		}
 		// This is to make sure that renames (which is an add followed by a delete) land in the same batch.
 		// Even if the batch is full, we allow a last delete to slip in, we do this by making sure that
 		// the batch ends with a non-delete, or that the last item in the batch is already a delete
 		if batch.Full() && (!fi.IsDeleted() || previousWasDelete) {
-			if err = batch.Flush(); err != nil {
-				return false
+			if err := batch.Flush(); err != nil {
+				return err
+			}
+			if time.Since(t0) > 5*time.Second {
+				// minor hack -- avoid very long running read transactions
+				// during index transmission, to help prevent excessive
+				// growth of database WAL file
+				break
 			}
 		}

@@ -307,6 +320,7 @@ func (s *indexHandler) sendIndexTo(ctx context.Context, fset *db.FileSet) error
 				"sequence": fi.SequenceNo(),
 				"start":    s.localPrevSequence + 1,
 			})
+			return errors.New("database misbehaved")
 		}

 		if f.Sequence > 0 && fi.SequenceNo() <= f.Sequence {
@@ -315,27 +329,17 @@ func (s *indexHandler) sendIndexTo(ctx context.Context, fset *db.FileSet) error
 				"start":    s.localPrevSequence + 1,
 				"previous": f.Sequence,
 			})
-			// Abort this round of index sending - the next one will pick
-			// up from the last successful one with the repeaired db.
-			defer func() {
-				if fixed, dbErr := fset.RepairSequence(); dbErr != nil {
-					l.Warnln("Failed repairing sequence entries:", dbErr)
-					panic("Failed repairing sequence entries")
-				} else {
-					s.evLogger.Log(events.Failure, "detected and repaired non-increasing sequence")
-					l.Infof("Repaired %v sequence entries in database", fixed)
-				}
-			}()
-			return false
+			return errors.New("database misbehaved")
 		}

 		f = fi
+		s.localPrevSequence = f.Sequence

 		// If this is a folder receiving encrypted files only, we
 		// mustn't ever send locally changed file infos. Those aren't
 		// encrypted and thus would be a protocol error at the remote.
 		if s.folderIsReceiveEncrypted && fi.IsReceiveOnlyChanged() {
-			return true
+			continue
 		}

 		f = prepareFileInfoForIndex(f)
@@ -343,23 +347,11 @@ func (s *indexHandler) sendIndexTo(ctx context.Context, fset *db.FileSet) error
 		previousWasDelete = f.IsDeleted()

 		batch.Append(f)
-		return true
-	})
-	if err != nil {
-		return err
 	}
-
 	if err := batch.Flush(); err != nil {
 		return err
 	}

-	// Use the sequence of the snapshot we iterated as a starting point for the
-	// next run. Previously we used the sequence of the last file we sent,
-	// however it's possible that a higher sequence exists, just doesn't need to
-	// be sent (e.g. in a receive-only folder, when a local change was
-	// reverted). No point trying to send nothing again.
-	s.localPrevSequence = snap.Sequence(protocol.LocalDeviceID)
-
 	return nil
 }

@@ -368,7 +360,6 @@ func (s *indexHandler) receive(fs []protocol.FileInfo, update bool, op string, p

 	s.cond.L.Lock()
 	paused := s.paused
-	fset := s.fset
 	runner := s.runner
 	s.cond.L.Unlock()

@@ -382,13 +373,19 @@ func (s *indexHandler) receive(fs []protocol.FileInfo, update bool, op string, p
 	s.downloads.Update(s.folder, makeForgetUpdate(fs))

 	if !update {
-		fset.Drop(deviceID)
+		if err := s.sdb.DropAllFiles(s.folder, deviceID); err != nil {
+			return err
+		}
 	}

 	l.Debugf("Received %d files for %s from %s, prevSeq=%d, lastSeq=%d", len(fs), s.folder, deviceID.Short(), prevSequence, lastSequence)

 	// Verify that the previous sequence number matches what we expected
-	if exp := fset.Sequence(deviceID); prevSequence > 0 && prevSequence != exp {
+	exp, err := s.sdb.GetDeviceSequence(s.folder, deviceID)
+	if err != nil {
+		return err
+	}
+	if prevSequence > 0 && prevSequence != exp {
 		s.logSequenceAnomaly("index update with unexpected sequence", map[string]any{
 			"prevSeq":      prevSequence,
 			"lastSeq":      lastSequence,
@@ -444,8 +441,13 @@ func (s *indexHandler) receive(fs []protocol.FileInfo, update bool, op string, p
 		})
 	}

-	fset.Update(deviceID, fs)
-	seq := fset.Sequence(deviceID)
+	if err := s.sdb.Update(s.folder, deviceID, fs); err != nil {
+		return err
+	}
+	seq, err := s.sdb.GetDeviceSequence(s.folder, deviceID)
+	if err != nil {
+		return err
+	}

 	// Check that the sequence we get back is what we put in...
 	if lastSequence > 0 && len(fs) > 0 && seq != lastSequence {
@@ -508,6 +510,7 @@ func (s *indexHandler) String() string {
 type indexHandlerRegistry struct {
 	evLogger      events.Logger
 	conn          protocol.Connection
+	sdb           db.DB
 	downloads     *deviceDownloadState
 	indexHandlers *serviceMap[string, *indexHandler]
 	startInfos    map[string]*clusterConfigDeviceInfo
@@ -517,14 +520,14 @@ type indexHandlerRegistry struct {

 type indexHandlerFolderState struct {
 	cfg    config.FolderConfiguration
-	fset   *db.FileSet
 	runner service
 }

-func newIndexHandlerRegistry(conn protocol.Connection, downloads *deviceDownloadState, evLogger events.Logger) *indexHandlerRegistry {
+func newIndexHandlerRegistry(conn protocol.Connection, sdb db.DB, downloads *deviceDownloadState, evLogger events.Logger) *indexHandlerRegistry {
 	r := &indexHandlerRegistry{
 		evLogger:      evLogger,
 		conn:          conn,
+		sdb:           sdb,
 		downloads:     downloads,
 		indexHandlers: newServiceMap[string, *indexHandler](evLogger),
 		startInfos:    make(map[string]*clusterConfigDeviceInfo),
@@ -544,15 +547,19 @@ func (r *indexHandlerRegistry) Serve(ctx context.Context) error {
 	return r.indexHandlers.Serve(ctx)
 }

-func (r *indexHandlerRegistry) startLocked(folder config.FolderConfiguration, fset *db.FileSet, runner service, startInfo *clusterConfigDeviceInfo) {
+func (r *indexHandlerRegistry) startLocked(folder config.FolderConfiguration, runner service, startInfo *clusterConfigDeviceInfo) error {
 	r.indexHandlers.RemoveAndWait(folder.ID, 0)
 	delete(r.startInfos, folder.ID)

-	is := newIndexHandler(r.conn, r.downloads, folder, fset, runner, startInfo, r.evLogger)
+	is, err := newIndexHandler(r.conn, r.downloads, folder, r.sdb, runner, startInfo, r.evLogger)
+	if err != nil {
+		return err
+	}
 	r.indexHandlers.Add(folder.ID, is)

 	// This new connection might help us get in sync.
 	runner.SchedulePull()
+	return nil
 }

 // AddIndexInfo starts an index handler for given folder, unless it is paused.
@@ -572,7 +579,7 @@ func (r *indexHandlerRegistry) AddIndexInfo(folder string, startInfo *clusterCon
 		r.startInfos[folder] = startInfo
 		return
 	}
-	r.startLocked(folderState.cfg, folderState.fset, folderState.runner, startInfo)
+	_ = r.startLocked(folderState.cfg, folderState.runner, startInfo) // XXX error handling...
 }

 // Remove stops a running index handler or removes one pending to be started.
@@ -612,7 +619,7 @@ func (r *indexHandlerRegistry) RemoveAllExcept(except map[string]remoteFolderSta
 // RegisterFolderState must be called whenever something about the folder
 // changes. The exception being if the folder is removed entirely, then call
 // Remove. The fset and runner arguments may be nil, if given folder is paused.
-func (r *indexHandlerRegistry) RegisterFolderState(folder config.FolderConfiguration, fset *db.FileSet, runner service) {
+func (r *indexHandlerRegistry) RegisterFolderState(folder config.FolderConfiguration, runner service) {
 	if !folder.SharedWith(r.conn.DeviceID()) {
 		r.Remove(folder.ID)
 		return
@@ -622,7 +629,7 @@ func (r *indexHandlerRegistry) RegisterFolderState(folder config.FolderConfigura
 	if folder.Paused {
 		r.folderPausedLocked(folder.ID)
 	} else {
-		r.folderRunningLocked(folder, fset, runner)
+		r.folderRunningLocked(folder, runner)
 	}
 	r.mut.Unlock()
 }
@@ -643,10 +650,9 @@ func (r *indexHandlerRegistry) folderPausedLocked(folder string) {
 // folderRunningLocked resumes an already running index handler or starts it, if it
 // was added while paused.
 // It is a noop if the folder isn't known.
-func (r *indexHandlerRegistry) folderRunningLocked(folder config.FolderConfiguration, fset *db.FileSet, runner service) {
+func (r *indexHandlerRegistry) folderRunningLocked(folder config.FolderConfiguration, runner service) {
 	r.folderStates[folder.ID] = &indexHandlerFolderState{
 		cfg:    folder,
-		fset:   fset,
 		runner: runner,
 	}

@@ -656,12 +662,12 @@ func (r *indexHandlerRegistry) folderRunningLocked(folder config.FolderConfigura
 			r.indexHandlers.RemoveAndWait(folder.ID, 0)
 			l.Debugf("Removed index handler for device %v and folder %v in resume", r.conn.DeviceID().Short(), folder.ID)
 		}
-		r.startLocked(folder, fset, runner, info)
+		_ = r.startLocked(folder, runner, info) // XXX error handling...
 		delete(r.startInfos, folder.ID)
 		l.Debugf("Started index handler for device %v and folder %v in resume", r.conn.DeviceID().Short(), folder.ID)
 	} else if isOk {
 		l.Debugf("Resuming index handler for device %v and folder %v", r.conn.DeviceID().Short(), folder)
-		is.resume(fset, runner)
+		is.resume(runner)
 	} else {
 		l.Debugf("Not resuming index handler for device %v and folder %v as none is paused and there is no start info", r.conn.DeviceID().Short(), folder.ID)
 	}