prevent MoveBucket from moving a bucket across two different db files

Signed-off-by: Mustafa Elbehery <melbeher@redhat.com>

bucket.go

@@ -343,10 +343,15 @@ func (b *Bucket) MoveBucket(key []byte, dstBucket *Bucket) (err error) {
 
 	if b.tx.db == nil || dstBucket.tx.db == nil {
 		return errors.ErrTxClosed
-	} else if !dstBucket.Writable() {
+	} else if !b.Writable() || !dstBucket.Writable() {
 		return errors.ErrTxNotWritable
 	}
 
+	if b.tx.db.Path() != dstBucket.tx.db.Path() || b.tx != dstBucket.tx {
+		lg.Errorf("The source and target buckets are not in the same db file, source bucket in %s and target bucket in %s", b.tx.db.Path(), dstBucket.tx.db.Path())
+		return errors.ErrDifferentDB
+	}
+
 	newKey := cloneBytes(key)
 
 	// Move cursor to correct position.

errors/errors.go

@@ -77,4 +77,8 @@ var (
 	// ErrSameBuckets is returned when trying to move a sub-bucket between
 	// source and target buckets, while source and target buckets are the same.
 	ErrSameBuckets = errors.New("the source and target are the same bucket")
+
+	// ErrDifferentDB is returned when trying to move a sub-bucket between
+	// source and target buckets, while source and target buckets are in different database files.
+	ErrDifferentDB = errors.New("the source and target buckets are in different database files")
 )

movebucket_test.go

@@ -143,7 +143,7 @@ func TestTx_MoveBucket(t *testing.T) {
 	for _, tc := range testCases {
 
 		t.Run(tc.name, func(*testing.T) {
-			db := btesting.MustCreateDBWithOption(t, &bbolt.Options{PageSize: 4096})
+			db := btesting.MustCreateDBWithOption(t, &bbolt.Options{PageSize: pageSize})
 
 			dumpBucketBeforeMoving := filepath.Join(t.TempDir(), "dbBeforeMove")
 			dumpBucketAfterMoving := filepath.Join(t.TempDir(), "dbAfterMove")
@@ -216,6 +216,151 @@ func TestTx_MoveBucket(t *testing.T) {
 	}
 }
 
+func TestBucket_MoveBucket_DiffDB(t *testing.T) {
+	srcBucketPath := []string{"sb1", "sb2"}
+	dstBucketPath := []string{"db1", "db2"}
+	bucketToMove := "bucketToMove"
+	expectedErr := errors.ErrDifferentDB
+
+	var srcBucket *bbolt.Bucket
+
+	t.Log("Creating source bucket and populate some data")
+	db1 := btesting.MustCreateDBWithOption(t, &bbolt.Options{PageSize: pageSize})
+	err := db1.Update(func(tx *bbolt.Tx) error {
+		srcBucket = prepareBuckets(t, tx, srcBucketPath...)
+		return nil
+	})
+	db1.MustClose()
+	require.NoError(t, err)
+
+	t.Log("Creating target bucket and populate some data")
+	db2 := btesting.MustCreateDBWithOption(t, &bbolt.Options{PageSize: pageSize})
+	err = db2.Update(func(tx *bbolt.Tx) error {
+		prepareBuckets(t, tx, dstBucketPath...)
+		return nil
+	})
+	db2.MustClose()
+	require.NoError(t, err)
+
+	t.Log("Re-opening source bucket from DB1")
+	srcDB, sErr := bbolt.Open(db1.Path(), 0600, bbolt.DefaultOptions)
+	require.NoError(t, sErr)
+	defer func() {
+		cErr := srcDB.Close()
+		require.NoError(t, cErr)
+	}()
+
+	t.Log("Reading source bucket in a separate RWTx")
+	sTx, sErr := srcDB.Begin(true)
+	require.NoError(t, sErr)
+	defer func() {
+		rErr := sTx.Rollback()
+		require.NoError(t, rErr)
+	}()
+	srcBucket = prepareBuckets(t, sTx, srcBucketPath...)
+
+	t.Log("Re-opening target bucket from DB2")
+	dstDB, dErr := bbolt.Open(db2.Path(), 0600, bbolt.DefaultOptions)
+	require.NoError(t, dErr)
+	defer func() {
+		cErr := dstDB.Close()
+		require.NoError(t, cErr)
+	}()
+
+	t.Log("Moving the sub-bucket in a separate RWTx")
+	err = dstDB.Update(func(tx *bbolt.Tx) error {
+		dstBucket := prepareBuckets(t, tx, dstBucketPath...)
+		mErr := srcBucket.MoveBucket([]byte(bucketToMove), dstBucket)
+		require.Equal(t, expectedErr, mErr)
+
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func TestBucket_MoveBucket_DiffTx(t *testing.T) {
+	testCases := []struct {
+		name          string
+		srcBucketPath []string
+		srcRWTx       bool
+		bucketToMove  string
+		dstBucketPath []string
+		dstRWTx       bool
+		expectedErr   error
+	}{
+		{
+			name:          "src is RWTx and target is RTx",
+			srcRWTx:       true,
+			srcBucketPath: []string{"sb1", "sb2"},
+			bucketToMove:  "bucketToMove",
+			dstBucketPath: []string{"db1", "db2"},
+			dstRWTx:       false,
+			expectedErr:   errors.ErrTxNotWritable,
+		},
+		{
+			name:          "src is RTx and target is RWTx",
+			srcRWTx:       false,
+			srcBucketPath: []string{"sb1", "sb2"},
+			bucketToMove:  "bucketToMove",
+			dstBucketPath: []string{"db1", "db2"},
+			dstRWTx:       true,
+			expectedErr:   errors.ErrTxNotWritable,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var srcBucket *bbolt.Bucket
+			var dstBucket *bbolt.Bucket
+
+			t.Log("Creating source and target buckets and populate some data")
+			db := btesting.MustCreateDBWithOption(t, &bbolt.Options{PageSize: pageSize})
+			err := db.Update(func(tx *bbolt.Tx) error {
+				srcBucket = prepareBuckets(t, tx, tc.srcBucketPath...)
+				dstBucket = prepareBuckets(t, tx, tc.dstBucketPath...)
+				return nil
+			})
+			db.MustClose()
+			require.NoError(t, err)
+
+			t.Log("Re-opening the database")
+			dB, dErr := bbolt.Open(db.Path(), 0600, bbolt.DefaultOptions)
+			require.NoError(t, dErr)
+			defer func() {
+				cErr := db.Close()
+				require.NoError(t, cErr)
+			}()
+
+			t.Log("Re-opening source bucket in a separate Tx")
+			sTx, sErr := dB.Begin(tc.srcRWTx)
+			require.NoError(t, sErr)
+			defer func() {
+				rErr := sTx.Rollback()
+				require.NoError(t, rErr)
+			}()
+			srcBucket = prepareBuckets(t, sTx, tc.srcBucketPath...)
+
+			t.Log("Re-opening target bucket in a separate Tx")
+			dTx, dErr := dB.Begin(tc.dstRWTx)
+			require.NoError(t, dErr)
+			defer func() {
+				rErr := dTx.Rollback()
+				require.NoError(t, rErr)
+			}()
+			dstBucket = prepareBuckets(t, dTx, tc.dstBucketPath...)
+
+			t.Log("Moving the sub-bucket")
+			err = dB.View(func(tx *bbolt.Tx) error {
+				mErr := srcBucket.MoveBucket([]byte(tc.bucketToMove), dstBucket)
+				require.Equal(t, tc.expectedErr, mErr)
+
+				return nil
+			})
+			require.NoError(t, err)
+		})
+	}
+}
+
 // prepareBuckets opens the bucket chain. For each bucket in the chain,
 // open it if existed, otherwise create it and populate sample data.
 func prepareBuckets(t testing.TB, tx *bbolt.Tx, buckets ...string) *bbolt.Bucket {