lib/db: Schema update to repair sequence index (ref #6304) (#6350)

One of the causes of "panic: database is closed" is that we try to send
summaries after it's been closed. Calculating summaries can take a long
time and if we have a lot of folders it's not unreasonable to think
that we might be stopped in this loop, so prepare to bail here.

* push

lib/db/schemaupdater.go

@@ -7,9 +7,11 @@
 package db
 
 import (
+	"bytes"
 	"fmt"
 	"strings"
 
+	"github.com/syncthing/syncthing/lib/db/backend"
 	"github.com/syncthing/syncthing/lib/protocol"
 )
 
@@ -23,11 +25,17 @@ import (
 //   6: v0.14.50
 //   7: v0.14.53
 //   8: v1.4.0
+//   9: v1.4.0
 const (
-	dbVersion             = 8
+	dbVersion             = 9
 	dbMinSyncthingVersion = "v1.4.0"
 )
 
+var (
+	errFolderIdxMissing = fmt.Errorf("folder db index missing")
+	errDeviceIdxMissing = fmt.Errorf("device db index missing")
+)
+
 type databaseDowngradeError struct {
 	minSyncthingVersion string
 }
@@ -80,7 +88,7 @@ func (db *schemaUpdater) updateSchema() error {
 		{5, db.updateSchemaTo5},
 		{6, db.updateSchema5to6},
 		{7, db.updateSchema6to7},
-		{8, db.updateSchema7to8},
+		{9, db.updateSchemato9},
 	}
 
 	for _, m := range migrations {
@@ -421,8 +429,9 @@ func (db *schemaUpdater) updateSchema6to7(_ int) error {
 	return t.Commit()
 }
 
-func (db *schemaUpdater) updateSchema7to8(_ int) error {
+func (db *schemaUpdater) updateSchemato9(prev int) error {
 	// Loads and rewrites all files with blocks, to deduplicate block lists.
+	// Checks for missing or incorrect sequence entries and rewrites those.
 
 	t, err := db.newReadWriteTransaction()
 	if err != nil {
@@ -430,15 +439,59 @@ func (db *schemaUpdater) updateSchema7to8(_ int) error {
 	}
 	defer t.close()
 
+	var sk []byte
 	it, err := t.NewPrefixIterator([]byte{KeyTypeDevice})
 	if err != nil {
 		return err
 	}
+	metas := make(map[string]*metadataTracker)
 	for it.Next() {
 		var fi protocol.FileInfo
 		if err := fi.Unmarshal(it.Value()); err != nil {
 			return err
 		}
+		device, ok := t.keyer.DeviceFromDeviceFileKey(it.Key())
+		if !ok {
+			return errDeviceIdxMissing
+		}
+		if bytes.Equal(device, protocol.LocalDeviceID[:]) {
+			folder, ok := t.keyer.FolderFromDeviceFileKey(it.Key())
+			if !ok {
+				return errFolderIdxMissing
+			}
+			if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
+				return err
+			}
+			switch dk, err := t.Get(sk); {
+			case err != nil:
+				if !backend.IsNotFound(err) {
+					return err
+				}
+				fallthrough
+			case !bytes.Equal(it.Key(), dk):
+				folderStr := string(folder)
+				meta, ok := metas[folderStr]
+				if !ok {
+					meta = loadMetadataTracker(db.Lowlevel, folderStr)
+					metas[folderStr] = meta
+				}
+				fi.Sequence = meta.nextLocalSeq()
+				if sk, err = t.keyer.GenerateSequenceKey(sk, folder, fi.Sequence); err != nil {
+					return err
+				}
+				if err := t.Put(sk, it.Key()); err != nil {
+					return err
+				}
+				if err := t.putFile(it.Key(), fi); err != nil {
+					return err
+				}
+				continue
+			}
+		}
+		if prev == 8 {
+			// The transition to 8 already did the changes below.
+			continue
+		}
 		if fi.Blocks == nil {
 			continue
 		}
@@ -451,6 +504,12 @@ func (db *schemaUpdater) updateSchema7to8(_ int) error {
 		return err
 	}
 
+	for folder, meta := range metas {
+		if err := meta.toDB(t, []byte(folder)); err != nil {
+			return err
+		}
+	}
+
 	db.recordTime(blockGCTimeKey)
 
 	return t.Commit()

lib/db/set.go

@@ -71,66 +71,68 @@ func init() {
 }
 
 func NewFileSet(folder string, fs fs.Filesystem, db *Lowlevel) *FileSet {
-	var s = &FileSet{
+	return &FileSet{
 		folder:      folder,
 		fs:          fs,
 		db:          db,
-		meta:        newMetadataTracker(),
+		meta:        loadMetadataTracker(db, folder),
 		updateMutex: sync.NewMutex(),
 	}
+}
 
-	recalc := func() *FileSet {
-		if err := s.recalcMeta(); backend.IsClosed(err) {
+func loadMetadataTracker(db *Lowlevel, folder string) *metadataTracker {
+	meta := newMetadataTracker()
+
+	recalc := func() *metadataTracker {
+		if err := recalcMeta(meta, db, folder); backend.IsClosed(err) {
 			return nil
 		} else if err != nil {
 			panic(err)
 		}
-		return s
+		return meta
 	}
 
-	if err := s.meta.fromDB(db, []byte(folder)); err != nil {
+	if err := meta.fromDB(db, []byte(folder)); err != nil {
 		l.Infof("No stored folder metadata for %q; recalculating", folder)
 		return recalc()
 	}
 
-	if metaOK := s.verifyLocalSequence(); !metaOK {
+	if metaOK := verifyLocalSequence(meta, db, folder); !metaOK {
 		l.Infof("Stored folder metadata for %q is out of date after crash; recalculating", folder)
 		return recalc()
 	}
 
-	if age := time.Since(s.meta.Created()); age > databaseRecheckInterval {
+	if age := time.Since(meta.Created()); age > databaseRecheckInterval {
 		l.Infof("Stored folder metadata for %q is %v old; recalculating", folder, age)
 		return recalc()
 	}
 
-	return s
+	return meta
 }
 
-func (s *FileSet) recalcMeta() error {
-	s.meta = newMetadataTracker()
-
-	if err := s.db.checkGlobals([]byte(s.folder), s.meta); err != nil {
+func recalcMeta(meta *metadataTracker, db *Lowlevel, folder string) error {
+	if err := db.checkGlobals([]byte(folder), meta); err != nil {
 		return err
 	}
 
-	t, err := s.db.newReadWriteTransaction()
+	t, err := db.newReadWriteTransaction()
 	if err != nil {
 		return err
 	}
 	defer t.close()
 
 	var deviceID protocol.DeviceID
-	err = t.withAllFolderTruncated([]byte(s.folder), func(device []byte, f FileInfoTruncated) bool {
+	err = t.withAllFolderTruncated([]byte(folder), func(device []byte, f FileInfoTruncated) bool {
 		copy(deviceID[:], device)
-		s.meta.addFile(deviceID, f)
+		meta.addFile(deviceID, f)
 		return true
 	})
 	if err != nil {
 		return err
 	}
 
-	s.meta.SetCreated()
-	if err := s.meta.toDB(t, []byte(s.folder)); err != nil {
+	meta.SetCreated()
+	if err := meta.toDB(t, []byte(folder)); err != nil {
 		return err
 	}
 	return t.Commit()
@@ -138,7 +140,7 @@ func (s *FileSet) recalcMeta() error {
 
 // Verify the local sequence number from actual sequence entries. Returns
 // true if it was all good, or false if a fixup was necessary.
-func (s *FileSet) verifyLocalSequence() bool {
+func verifyLocalSequence(meta *metadataTracker, db *Lowlevel, folder string) bool {
 	// Walk the sequence index from the current (supposedly) highest
 	// sequence number and raise the alarm if we get anything. This recovers
 	// from the occasion where we have written sequence entries to disk but
@@ -149,15 +151,20 @@ func (s *FileSet) verifyLocalSequence() bool {
 	// number than we've actually seen and receive some duplicate updates
 	// and then be in sync again.
 
-	curSeq := s.meta.Sequence(protocol.LocalDeviceID)
+	curSeq := meta.Sequence(protocol.LocalDeviceID)
 
-	snap := s.Snapshot()
+	t, err := db.newReadOnlyTransaction()
+	if err != nil {
+		panic(err)
+	}
 	ok := true
-	snap.WithHaveSequence(curSeq+1, func(fi FileIntf) bool {
+	if err := t.withHaveSequence([]byte(folder), curSeq+1, func(fi FileIntf) bool {
 		ok = false // we got something, which we should not have
 		return false
-	})
-	snap.Release()
+	}); err != nil && !backend.IsClosed(err) {
+		panic(err)
+	}
+	t.close()
 
 	return ok
 }

lib/db/structs.go

@@ -129,27 +129,36 @@ func (f FileInfoTruncated) FileModifiedBy() protocol.ShortID {
 }
 
 func (f FileInfoTruncated) ConvertToIgnoredFileInfo(by protocol.ShortID) protocol.FileInfo {
-	return protocol.FileInfo{
-		Name:         f.Name,
-		Type:         f.Type,
-		ModifiedS:    f.ModifiedS,
-		ModifiedNs:   f.ModifiedNs,
-		ModifiedBy:   by,
-		Version:      f.Version,
-		RawBlockSize: f.RawBlockSize,
-		LocalFlags:   protocol.FlagLocalIgnored,
-	}
+	file := f.copyToFileInfo()
+	file.SetIgnored(by)
+	return file
 }
 
-func (f FileInfoTruncated) ConvertToDeletedFileInfo(by protocol.ShortID, localFlags uint32) protocol.FileInfo {
+func (f FileInfoTruncated) ConvertToDeletedFileInfo(by protocol.ShortID) protocol.FileInfo {
+	file := f.copyToFileInfo()
+	file.SetDeleted(by)
+	return file
+}
+
+// copyToFileInfo just copies all members of FileInfoTruncated to protocol.FileInfo
+func (f FileInfoTruncated) copyToFileInfo() protocol.FileInfo {
 	return protocol.FileInfo{
-		Name:       f.Name,
-		Type:       f.Type,
-		ModifiedS:  time.Now().Unix(),
-		ModifiedBy: by,
-		Deleted:    true,
-		Version:    f.Version.Update(by),
-		LocalFlags: localFlags,
+		Name:          f.Name,
+		Size:          f.Size,
+		ModifiedS:     f.ModifiedS,
+		ModifiedBy:    f.ModifiedBy,
+		Version:       f.Version,
+		Sequence:      f.Sequence,
+		SymlinkTarget: f.SymlinkTarget,
+		BlocksHash:    f.BlocksHash,
+		Type:          f.Type,
+		Permissions:   f.Permissions,
+		ModifiedNs:    f.ModifiedNs,
+		RawBlockSize:  f.RawBlockSize,
+		LocalFlags:    f.LocalFlags,
+		Deleted:       f.Deleted,
+		RawInvalid:    f.RawInvalid,
+		NoPermissions: f.NoPermissions,
 	}
 }
 

lib/db/transactions.go

@@ -468,6 +468,9 @@ func (t readWriteTransaction) putFile(key []byte, fi protocol.FileInfo) error {
 		} else if err != nil {
 			return err
 		}
+	} else if fi.BlocksHash != nil {
+		l.Warnln("Blocks is nil, but BlocksHash is not for file", fi)
+		panic("Blocks is nil, but BlocksHash is not")
 	}
 
 	fi.Blocks = nil
@@ -755,10 +758,7 @@ func (t *readWriteTransaction) withAllFolderTruncated(folder []byte, fn func(dev
 			return nil
 		}
 	}
-	if err := dbi.Error(); err != nil {
-		return err
-	}
-	return t.Commit()
+	return dbi.Error()
 }
 
 type marshaller interface {

lib/model/folder.go

@@ -532,7 +532,8 @@ func (f *folder) scanSubdirs(subDirs []string) error {
 					}
 					return true
 				}
-				nf := file.ConvertToDeletedFileInfo(f.shortID, f.localFlags)
+				nf := file.ConvertToDeletedFileInfo(f.shortID)
+				nf.LocalFlags = f.localFlags
 				if file.ShouldConflict() {
 					// We do not want to override the global version with
 					// the deleted file. Setting to an empty version makes

lib/model/folder_recvonly.go

@@ -104,14 +104,8 @@ func (f *receiveOnlyFolder) Revert() {
 				return true // continue
 			}
 
-			fi = protocol.FileInfo{
-				Name:       fi.Name,
-				Type:       fi.Type,
-				ModifiedS:  time.Now().Unix(),
-				ModifiedBy: f.shortID,
-				Deleted:    true,
-				Version:    protocol.Vector{}, // if this file ever resurfaces anywhere we want our delete to be strictly older
-			}
+			fi.SetDeleted(f.shortID)
+			fi.Version = protocol.Vector{} // if this file ever resurfaces anywhere we want our delete to be strictly older
 		} else {
 			// Revert means to throw away our local changes. We reset the
 			// version to the empty vector, which is strictly older than any

lib/model/folder_sendonly.go

@@ -118,10 +118,7 @@ func (f *sendOnlyFolder) Override() {
 		}
 		if !ok || have.Name != need.Name {
 			// We are missing the file
-			need.Deleted = true
-			need.Blocks = nil
-			need.Version = need.Version.Update(f.shortID)
-			need.Size = 0
+			need.SetDeleted(f.shortID)
 		} else {
 			// We have the file, replace with our version
 			have.Version = have.Version.Merge(need.Version).Update(f.shortID)

lib/model/folder_summary.go

@@ -270,7 +270,12 @@ func (c *folderSummaryService) calculateSummaries(ctx context.Context) {
 		case <-pump.C:
 			t0 := time.Now()
 			for _, folder := range c.foldersToHandle() {
-				c.sendSummary(folder)
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+				c.sendSummary(ctx, folder)
 			}
 
 			// We don't want to spend all our time calculating summaries. Lets
@@ -280,7 +285,7 @@ func (c *folderSummaryService) calculateSummaries(ctx context.Context) {
 			pump.Reset(wait)
 
 		case folder := <-c.immediate:
-			c.sendSummary(folder)
+			c.sendSummary(ctx, folder)
 
 		case <-ctx.Done():
 			return
@@ -313,7 +318,7 @@ func (c *folderSummaryService) foldersToHandle() []string {
 }
 
 // sendSummary send the summary events for a single folder
-func (c *folderSummaryService) sendSummary(folder string) {
+func (c *folderSummaryService) sendSummary(ctx context.Context, folder string) {
 	// The folder summary contains how many bytes, files etc
 	// are in the folder and how in sync we are.
 	data, err := c.Summary(folder)
@@ -326,6 +331,12 @@ func (c *folderSummaryService) sendSummary(folder string) {
 	})
 
 	for _, devCfg := range c.cfg.Folders()[folder].Devices {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
 		if devCfg.DeviceID.Equals(c.id) {
 			// We already know about ourselves.
 			continue

lib/protocol/bep_extensions.go

@@ -266,24 +266,36 @@ func BlocksEqual(a, b []BlockInfo) bool {
 }
 
 func (f *FileInfo) SetMustRescan(by ShortID) {
-	f.LocalFlags = FlagLocalMustRescan
-	f.ModifiedBy = by
-	f.Blocks = nil
-	f.Sequence = 0
+	f.setLocalFlags(by, FlagLocalMustRescan)
 }
 
 func (f *FileInfo) SetIgnored(by ShortID) {
-	f.LocalFlags = FlagLocalIgnored
-	f.ModifiedBy = by
-	f.Blocks = nil
-	f.Sequence = 0
+	f.setLocalFlags(by, FlagLocalIgnored)
 }
 
 func (f *FileInfo) SetUnsupported(by ShortID) {
-	f.LocalFlags = FlagLocalUnsupported
+	f.setLocalFlags(by, FlagLocalUnsupported)
+}
+
+func (f *FileInfo) SetDeleted(by ShortID) {
 	f.ModifiedBy = by
+	f.Deleted = true
+	f.Version = f.Version.Update(by)
+	f.ModifiedS = time.Now().Unix()
+	f.setNoContent()
+}
+
+func (f *FileInfo) setLocalFlags(by ShortID, flags uint32) {
+	f.RawInvalid = false
+	f.LocalFlags = flags
+	f.ModifiedBy = by
+	f.setNoContent()
+}
+
+func (f *FileInfo) setNoContent() {
 	f.Blocks = nil
-	f.Sequence = 0
+	f.BlocksHash = nil
+	f.Size = 0
 }
 
 func (b BlockInfo) String() string {