fix(enterprise): Set version correctly post marshalling during restore

posting/list.go

@@ -918,6 +918,8 @@ func (out *rollupOutput) marshalPostingListPart(alloc *z.Allocator,
 	return kv, nil
 }
 
+// MarshalPostingList returns a KV with the marshalled posting list. The caller
+// SHOULD SET the Key and Version for the returned KV.
 func MarshalPostingList(plist *pb.PostingList, alloc *z.Allocator) *bpb.KV {
 	kv := y.NewKV(alloc)
 	if isPlistEmpty(plist) {

systest/backup/filesystem/backup_test.go

@@ -40,19 +40,87 @@ import (
 )
 
 var (
-	copyBackupDir = "./data/backups_copy"
-	restoreDir    = "./data/restore"
-	testDirs      = []string{restoreDir}
-
-	alphaBackupDir = "/data/backups"
-
+	copyBackupDir   = "./data/backups_copy"
+	restoreDir      = "./data/restore"
+	testDirs        = []string{restoreDir, exporterDir}
+	exporterDir     = "data/exporter"
+	alphaBackupDir  = "/data/backups"
+	oldBackupDir    = "/data/to_restore"
 	alphaContainers = []string{
 		"alpha1",
 		"alpha2",
 		"alpha3",
 	}
 )
 
+func sendRestoreRequest(t *testing.T, location string) int {
+	if location == "" {
+		location = "/data/backup"
+	}
+	params := testutil.GraphQLParams{
+		Query: `mutation restore($location: String!) {
+			restore(input: {location: $location}) {
+				code
+				message
+				restoreId
+			}
+		}`,
+		Variables: map[string]interface{}{
+			"location": location,
+		},
+	}
+	resp := testutil.MakeGQLRequest(t, &params)
+	resp.RequireNoGraphQLErrors(t)
+
+	var restoreResp struct {
+		Restore struct {
+			Code      string
+			Message   string
+			RestoreId int
+		}
+	}
+
+	require.NoError(t, json.Unmarshal(resp.Data, &restoreResp))
+	require.Equal(t, restoreResp.Restore.Code, "Success")
+	require.Greater(t, restoreResp.Restore.RestoreId, 0)
+	return restoreResp.Restore.RestoreId
+}
+
+// This test takes a backup and then restores an old backup in a cluster incrementally.
+// Next, cleans up the cluster and tries restoring the backups above.
+func TestBackupOfOldRestore(t *testing.T) {
+	dirSetup(t)
+	copyOldBackupDir(t)
+
+	dg, err := testutil.DgraphClient(testutil.SockAddr)
+	x.Check(err)
+	ctx := context.Background()
+
+	require.NoError(t, dg.Alter(ctx, &api.Operation{DropAll: true}))
+	time.Sleep(2 * time.Second)
+
+	_ = runBackup(t, 3, 1)
+
+	_ = sendRestoreRequest(t, oldBackupDir)
+	time.Sleep(3 * time.Second)
+
+	resp, err := dg.NewTxn().Query(context.Background(), `{ authors(func: has(Author.name)) { count(uid) } }`)
+	x.Check(err)
+	require.JSONEq(t, "{\"authors\":[{\"count\":1}]}", string(resp.Json))
+
+	_ = runBackup(t, 6, 2)
+
+	// Clean the cluster and try restoring the backups created above.
+	require.NoError(t, dg.Alter(ctx, &api.Operation{DropAll: true}))
+	time.Sleep(2 * time.Second)
+	_ = sendRestoreRequest(t, alphaBackupDir)
+	time.Sleep(3 * time.Second)
+
+	resp, err = dg.NewTxn().Query(context.Background(), `{ authors(func: has(Author.name)) { count(uid) } }`)
+	x.Check(err)
+	require.JSONEq(t, "{\"authors\":[{\"count\":1}]}", string(resp.Json))
+}
+
 func TestBackupFilesystem(t *testing.T) {
 	conn, err := grpc.Dial(testutil.SockAddr, grpc.WithInsecure())
 	require.NoError(t, err)
@@ -330,7 +398,7 @@ func runBackupInternal(t *testing.T, forceFull bool, numExpectedFiles,
 	copyToLocalFs(t)
 
 	files := x.WalkPathFunc(copyBackupDir, func(path string, isdir bool) bool {
-		return !isdir && strings.HasSuffix(path, ".backup")
+		return !isdir && strings.HasSuffix(path, ".backup") && strings.HasPrefix(path, "data/backups_copy/dgraph.")
 	})
 	require.Equal(t, numExpectedFiles, len(files))
 
@@ -340,7 +408,7 @@ func runBackupInternal(t *testing.T, forceFull bool, numExpectedFiles,
 	require.Equal(t, numExpectedDirs, len(dirs))
 
 	manifests := x.WalkPathFunc(copyBackupDir, func(path string, isdir bool) bool {
-		return !isdir && strings.Contains(path, "manifest.json")
+		return !isdir && strings.Contains(path, "manifest.json") && strings.HasPrefix(path, "data/backups_copy/dgraph.")
 	})
 	require.Equal(t, numExpectedDirs, len(manifests))
 
@@ -403,6 +471,9 @@ func dirCleanup(t *testing.T) {
 	if err := os.RemoveAll(restoreDir); err != nil {
 		t.Fatalf("Error removing directory: %s", err.Error())
 	}
+	if err := os.RemoveAll(exporterDir); err != nil {
+		t.Fatalf("Error removing directory: %s", err.Error())
+	}
 	if err := os.RemoveAll(copyBackupDir); err != nil {
 		t.Fatalf("Error removing directory: %s", err.Error())
 	}
@@ -413,6 +484,14 @@ func dirCleanup(t *testing.T) {
 	}
 }
 
+func copyOldBackupDir(t *testing.T) {
+	destPath := testutil.DockerPrefix + "_alpha1_1:/data"
+	srchPath := "." + oldBackupDir
+	if err := testutil.DockerCp(srchPath, destPath); err != nil {
+		t.Fatalf("Error copying files from docker container: %s", err.Error())
+	}
+}
+
 func copyToLocalFs(t *testing.T) {
 	// The original backup files are not accessible because docker creates all files in
 	// the shared volume as the root user. This restriction is circumvented by using

worker/restore.go

@@ -183,10 +183,14 @@ func loadFromBackup(db *badger.DB, r io.Reader, restoreTs uint64, preds predicat
 					// part without rolling the key first. This part is here for backwards
 					// compatibility. New backups are not affected because there was a change
 					// to roll up lists into a single one.
-					kv := posting.MarshalPostingList(pl, nil)
+					newKv := posting.MarshalPostingList(pl, nil)
 					codec.FreePack(pl.Pack)
-					kv.Key = restoreKey
-					if err := loader.Set(kv); err != nil {
+					newKv.Key = restoreKey
+					// Use the version of the KV before we marshalled the
+					// posting list. The MarshalPostingList function returns KV
+					// with a zero version.
+					newKv.Version = kv.Version
+					if err := loader.Set(newKv); err != nil {
 						return 0, err
 					}
 				} else {