fix crash loop on missing manifest tables (#2198)

Please see
https://discuss.hypermode.com/t/manifest-removes-non-existing-table/19882

We are experiencing an error where dgraph tries to remove a table that is
missing. I assume that given it has already been removed, it should not
trigger a crash loop in compaction. Instead i sense check that the table
is indeed removed from all levels, and return without an error code.

Author: RJKeevil

badger/cmd/info.go

@@ -16,7 +16,7 @@ import (
 	"strings"
 	"time"
 
-	humanize "github.com/dustin/go-humanize"
+	"github.com/dustin/go-humanize"
 	"github.com/spf13/cobra"
 
 	"github.com/dgraph-io/badger/v4"
@@ -109,7 +109,7 @@ func handleInfo(cmd *cobra.Command, args []string) error {
 		return nil
 	}
 
-	if err := printInfo(sstDir, vlogDir); err != nil {
+	if err := printInfo(sstDir, vlogDir, bopt); err != nil {
 		return y.Wrap(err, "failed to print information in MANIFEST file")
 	}
 
@@ -320,7 +320,7 @@ func readDir(dir string) ([]fs.FileInfo, error) {
 	return infos, err
 }
 
-func printInfo(dir, valueDir string) error {
+func printInfo(dir, valueDir string, bopt badger.Options) error {
 	if dir == "" {
 		return fmt.Errorf("--dir not supplied")
 	}
@@ -336,7 +336,7 @@ func printInfo(dir, valueDir string) error {
 			fp.Close()
 		}
 	}()
-	manifest, truncOffset, err := badger.ReplayManifestFile(fp, opt.externalMagicVersion)
+	manifest, truncOffset, err := badger.ReplayManifestFile(fp, opt.externalMagicVersion, bopt)
 	if err != nil {
 		return err
 	}

db2_test.go

@@ -427,24 +427,24 @@ func TestCompactionFilePicking(t *testing.T) {
 	for i := 1; i <= 10; i++ {
 		// Each table has difference of 1 between smallest and largest key.
 		tab := createTableWithRange(t, db, 2*i-1, 2*i)
-		addToManifest(t, db, tab, 3)
+		addToManifest(t, db, tab, 3, db.opt)
 		require.NoError(t, l3.replaceTables([]*table.Table{}, []*table.Table{tab}))
 	}
 
 	l2 := db.lc.levels[2]
 	// First table has keys 1 and 4.
 	tab := createTableWithRange(t, db, 1, 4)
-	addToManifest(t, db, tab, 2)
+	addToManifest(t, db, tab, 2, db.opt)
 	require.NoError(t, l2.replaceTables([]*table.Table{}, []*table.Table{tab}))
 
 	// Second table has keys 5 and 12.
 	tab = createTableWithRange(t, db, 5, 12)
-	addToManifest(t, db, tab, 2)
+	addToManifest(t, db, tab, 2, db.opt)
 	require.NoError(t, l2.replaceTables([]*table.Table{}, []*table.Table{tab}))
 
 	// Third table has keys 13 and 18.
 	tab = createTableWithRange(t, db, 13, 18)
-	addToManifest(t, db, tab, 2)
+	addToManifest(t, db, tab, 2, db.opt)
 	require.NoError(t, l2.replaceTables([]*table.Table{}, []*table.Table{tab}))
 
 	cdef := &compactDef{
@@ -479,14 +479,14 @@ func TestCompactionFilePicking(t *testing.T) {
 }
 
 // addToManifest function is used in TestCompactionFilePicking. It adds table to db manifest.
-func addToManifest(t *testing.T, db *DB, tab *table.Table, level uint32) {
+func addToManifest(t *testing.T, db *DB, tab *table.Table, level uint32, opt Options) {
 	change := &pb.ManifestChange{
 		Id:          tab.ID(),
 		Op:          pb.ManifestChange_CREATE,
 		Level:       level,
 		Compression: uint32(tab.CompressionType()),
 	}
-	require.NoError(t, db.manifest.addChanges([]*pb.ManifestChange{change}),
+	require.NoError(t, db.manifest.addChanges([]*pb.ManifestChange{change}, opt),
 		"unable to add to manifest")
 }
 
@@ -679,7 +679,7 @@ func TestWindowsDataLoss(t *testing.T) {
 			v := []byte("barValuebarValuebarValuebarValuebarValue")
 			require.Greater(t, len(v), db.valueThreshold())
 
-			//32 bytes length and now it's not working
+			// 32 bytes length and now it's not working
 			err := txn.Set(key, v)
 			require.NoError(t, err)
 			keyList = append(keyList, key)

levels.go

@@ -226,7 +226,7 @@ func (s *levelsController) dropTree() (int, error) {
 		}
 	}
 	changeSet := pb.ManifestChangeSet{Changes: changes}
-	if err := s.kv.manifest.addChanges(changeSet.Changes); err != nil {
+	if err := s.kv.manifest.addChanges(changeSet.Changes, s.kv.opt); err != nil {
 		return 0, err
 	}
 
@@ -1437,7 +1437,7 @@ func (s *levelsController) runCompactDef(id, l int, cd compactDef) (err error) {
 	changeSet := buildChangeSet(&cd, newTables)
 
 	// We write to the manifest _before_ we delete files (and after we created files)
-	if err := s.kv.manifest.addChanges(changeSet.Changes); err != nil {
+	if err := s.kv.manifest.addChanges(changeSet.Changes, s.kv.opt); err != nil {
 		return err
 	}
 
@@ -1566,7 +1566,7 @@ func (s *levelsController) addLevel0Table(t *table.Table) error {
 		// deletes the table.)
 		err := s.kv.manifest.addChanges([]*pb.ManifestChange{
 			newCreateChange(t.ID(), 0, t.KeyID(), t.CompressionType()),
-		})
+		}, s.kv.opt)
 		if err != nil {
 			return err
 		}

levels_test.go

@@ -45,7 +45,7 @@ func createAndOpen(db *DB, td []keyValVersion, level int) {
 	}
 	if err := db.manifest.addChanges([]*pb.ManifestChange{
 		newCreateChange(tab.ID(), level, 0, tab.CompressionType()),
-	}); err != nil {
+	}, db.opt); err != nil {
 		panic(err)
 	}
 	db.lc.levels[level].Lock()
@@ -1193,7 +1193,7 @@ func TestFillTableCleanup(t *testing.T) {
 				tab := buildTable(i)
 				require.NoError(t, db.manifest.addChanges([]*pb.ManifestChange{
 					newCreateChange(tab.ID(), level, 0, tab.CompressionType()),
-				}))
+				}, db.opt))
 				tab.CreatedAt = time.Now().Add(-10 * time.Hour)
 				// Add table to the given level.
 				lh.addTable(tab)
@@ -1270,7 +1270,7 @@ func TestStaleDataCleanup(t *testing.T) {
 			tab := buildStaleTable(i)
 			require.NoError(t, db.manifest.addChanges([]*pb.ManifestChange{
 				newCreateChange(tab.ID(), level, 0, tab.CompressionType()),
-			}))
+			}, db.opt))
 			tab.CreatedAt = time.Now().Add(-10 * time.Hour)
 			// Add table to the given level.
 			lh.addTable(tab)

manifest.go

@@ -105,10 +105,10 @@ func (m *Manifest) asChanges() []*pb.ManifestChange {
 	return changes
 }
 
-func (m *Manifest) clone() Manifest {
+func (m *Manifest) clone(opt Options) Manifest {
 	changeSet := pb.ManifestChangeSet{Changes: m.asChanges()}
 	ret := createManifest()
-	y.Check(applyChangeSet(&ret, &changeSet))
+	y.Check(applyChangeSet(&ret, &changeSet, opt))
 	return ret
 }
 
@@ -120,11 +120,11 @@ func openOrCreateManifestFile(opt Options) (
 		return &manifestFile{inMemory: true}, Manifest{}, nil
 	}
 	return helpOpenOrCreateManifestFile(opt.Dir, opt.ReadOnly, opt.ExternalMagicVersion,
-		manifestDeletionsRewriteThreshold)
+		manifestDeletionsRewriteThreshold, opt)
 }
 
 func helpOpenOrCreateManifestFile(dir string, readOnly bool, extMagic uint16,
-	deletionsThreshold int) (*manifestFile, Manifest, error) {
+	deletionsThreshold int, opt Options) (*manifestFile, Manifest, error) {
 
 	path := filepath.Join(dir, ManifestFilename)
 	var flags y.Flags
@@ -149,13 +149,13 @@ func helpOpenOrCreateManifestFile(dir string, readOnly bool, extMagic uint16,
 			fp:                        fp,
 			directory:                 dir,
 			externalMagic:             extMagic,
-			manifest:                  m.clone(),
+			manifest:                  m.clone(opt),
 			deletionsRewriteThreshold: deletionsThreshold,
 		}
 		return mf, m, nil
 	}
 
-	manifest, truncOffset, err := ReplayManifestFile(fp, extMagic)
+	manifest, truncOffset, err := ReplayManifestFile(fp, extMagic, opt)
 	if err != nil {
 		_ = fp.Close()
 		return nil, Manifest{}, err
@@ -177,7 +177,7 @@ func helpOpenOrCreateManifestFile(dir string, readOnly bool, extMagic uint16,
 		fp:                        fp,
 		directory:                 dir,
 		externalMagic:             extMagic,
-		manifest:                  manifest.clone(),
+		manifest:                  manifest.clone(opt),
 		deletionsRewriteThreshold: deletionsThreshold,
 	}
 	return mf, manifest, nil
@@ -194,7 +194,7 @@ func (mf *manifestFile) close() error {
 // we replay the MANIFEST file, we'll either replay all the changes or none of them.  (The truth of
 // this depends on the filesystem -- some might append garbage data if a system crash happens at
 // the wrong time.)
-func (mf *manifestFile) addChanges(changesParam []*pb.ManifestChange) error {
+func (mf *manifestFile) addChanges(changesParam []*pb.ManifestChange, opt Options) error {
 	if mf.inMemory {
 		return nil
 	}
@@ -207,7 +207,7 @@ func (mf *manifestFile) addChanges(changesParam []*pb.ManifestChange) error {
 	// Maybe we could use O_APPEND instead (on certain file systems)
 	mf.appendLock.Lock()
 	defer mf.appendLock.Unlock()
-	if err := applyChangeSet(&mf.manifest, &changes); err != nil {
+	if err := applyChangeSet(&mf.manifest, &changes, opt); err != nil {
 		return err
 	}
 	// Rewrite manifest if it'd shrink by 1/10 and it's big enough to care
@@ -350,7 +350,7 @@ var (
 // Also, returns the last offset after a completely read manifest entry -- the file must be
 // truncated at that point before further appends are made (if there is a partial entry after
 // that).  In normal conditions, truncOffset is the file size.
-func ReplayManifestFile(fp *os.File, extMagic uint16) (Manifest, int64, error) {
+func ReplayManifestFile(fp *os.File, extMagic uint16, opt Options) (Manifest, int64, error) {
 	r := countingReader{wrapped: bufio.NewReader(fp)}
 
 	var magicBuf [8]byte
@@ -418,15 +418,15 @@ func ReplayManifestFile(fp *os.File, extMagic uint16) (Manifest, int64, error) {
 			return Manifest{}, 0, err
 		}
 
-		if err := applyChangeSet(&build, &changeSet); err != nil {
+		if err := applyChangeSet(&build, &changeSet, opt); err != nil {
 			return Manifest{}, 0, err
 		}
 	}
 
 	return build, offset, nil
 }
 
-func applyManifestChange(build *Manifest, tc *pb.ManifestChange) error {
+func applyManifestChange(build *Manifest, tc *pb.ManifestChange, opt Options) error {
 	switch tc.Op {
 	case pb.ManifestChange_CREATE:
 		if _, ok := build.Tables[tc.Id]; ok {
@@ -445,10 +445,14 @@ func applyManifestChange(build *Manifest, tc *pb.ManifestChange) error {
 	case pb.ManifestChange_DELETE:
 		tm, ok := build.Tables[tc.Id]
 		if !ok {
-			return fmt.Errorf("MANIFEST removes non-existing table %d", tc.Id)
+			opt.Warningf("MANIFEST delete: table %d has already been removed", tc.Id)
+			for _, level := range build.Levels {
+				delete(level.Tables, tc.Id)
+			}
+		} else {
+			delete(build.Levels[tm.Level].Tables, tc.Id)
+			delete(build.Tables, tc.Id)
 		}
-		delete(build.Levels[tm.Level].Tables, tc.Id)
-		delete(build.Tables, tc.Id)
 		build.Deletions++
 	default:
 		return fmt.Errorf("MANIFEST file has invalid manifestChange op")
@@ -458,9 +462,9 @@ func applyManifestChange(build *Manifest, tc *pb.ManifestChange) error {
 
 // This is not a "recoverable" error -- opening the KV store fails because the MANIFEST file is
 // just plain broken.
-func applyChangeSet(build *Manifest, changeSet *pb.ManifestChangeSet) error {
+func applyChangeSet(build *Manifest, changeSet *pb.ManifestChangeSet, opt Options) error {
 	for _, change := range changeSet.Changes {
-		if err := applyManifestChange(build, change); err != nil {
+		if err := applyManifestChange(build, change, opt); err != nil {
 			return err
 		}
 	}

manifest_test.go

@@ -196,9 +196,18 @@ func TestOverlappingKeyRangeError(t *testing.T) {
 func TestManifestRewrite(t *testing.T) {
 	dir, err := os.MkdirTemp("", "badger-test")
 	require.NoError(t, err)
-	defer removeDir(dir)
+
+	db, err := Open(DefaultOptions(dir))
+	require.NoError(t, err, "error while opening db")
+
+	defer func() {
+		require.NoError(t, db.Close())
+		removeDir(dir)
+	}()
+	
 	deletionsThreshold := 10
-	mf, m, err := helpOpenOrCreateManifestFile(dir, false, 0, deletionsThreshold)
+
+	mf, m, err := helpOpenOrCreateManifestFile(dir, false, 0, deletionsThreshold, db.opt)
 	defer func() {
 		if mf != nil {
 			mf.close()
@@ -210,21 +219,21 @@ func TestManifestRewrite(t *testing.T) {
 
 	err = mf.addChanges([]*pb.ManifestChange{
 		newCreateChange(0, 0, 0, 0),
-	})
+	}, db.opt)
 	require.NoError(t, err)
 
 	for i := uint64(0); i < uint64(deletionsThreshold*3); i++ {
 		ch := []*pb.ManifestChange{
 			newCreateChange(i+1, 0, 0, 0),
 			newDeleteChange(i),
 		}
-		err := mf.addChanges(ch)
+		err := mf.addChanges(ch, db.opt)
 		require.NoError(t, err)
 	}
 	err = mf.close()
 	require.NoError(t, err)
 	mf = nil
-	mf, m, err = helpOpenOrCreateManifestFile(dir, false, 0, deletionsThreshold)
+	mf, m, err = helpOpenOrCreateManifestFile(dir, false, 0, deletionsThreshold, db.opt)
 	require.NoError(t, err)
 	require.Equal(t, map[uint64]TableManifest{
 		uint64(deletionsThreshold * 3): {Level: 0},
@@ -236,14 +245,20 @@ func TestConcurrentManifestCompaction(t *testing.T) {
 	require.NoError(t, err)
 	defer removeDir(dir)
 
+	db, err := Open(DefaultOptions(dir))
+	require.NoError(t, err, "error while opening db")
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
 	// overwrite the sync function to make this race condition easily reproducible
 	syncFunc = func(f *os.File) error {
 		// effectively making the Sync() take around 1s makes this reproduce every time
 		time.Sleep(1 * time.Second)
 		return f.Sync()
 	}
 
-	mf, _, err := helpOpenOrCreateManifestFile(dir, false, 0, 0)
+	mf, _, err := helpOpenOrCreateManifestFile(dir, false, 0, 0, db.opt)
 	require.NoError(t, err)
 
 	cs := &pb.ManifestChangeSet{}
@@ -261,7 +276,7 @@ func TestConcurrentManifestCompaction(t *testing.T) {
 	for i := 0; i < n; i++ {
 		go func() {
 			defer wg.Done()
-			require.NoError(t, mf.addChanges(cs.Changes))
+			require.NoError(t, mf.addChanges(cs.Changes, db.opt))
 		}()
 	}
 	wg.Wait()

stream_writer.go

@@ -10,7 +10,7 @@ import (
 	"fmt"
 	"sync"
 
-	humanize "github.com/dustin/go-humanize"
+	"github.com/dustin/go-humanize"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dgraph-io/badger/v4/pb"
@@ -504,7 +504,7 @@ func (w *sortedWriter) createTable(builder *table.Builder) error {
 		Level:       uint32(lhandler.level),
 		Compression: uint32(tbl.CompressionType()),
 	}
-	if err := w.db.manifest.addChanges([]*pb.ManifestChange{change}); err != nil {
+	if err := w.db.manifest.addChanges([]*pb.ManifestChange{change}, w.db.opt); err != nil {
 		return err
 	}