From 4e7d49791b221b7115595f4a72a3dd85fa2e46d6 Mon Sep 17 00:00:00 2001
From: Alex Vanin <alexey@nspcc.ru>
Date: Fri, 27 Nov 2020 11:02:10 +0300
Subject: [PATCH] [#199] Index parent first in metabase

With exist check we should index parent first, because
as soon as child will be added to metabase, exist on
parent will return true even if it was not indexed yet.

Also this commit makes one db.Update instead of two for
parent and child.

Signed-off-by: Alex Vanin <alexey@nspcc.ru>
---
 pkg/local_object_storage/metabase/v2/put.go | 94 +++++++++++----------
 1 file changed, 48 insertions(+), 46 deletions(-)

diff --git a/pkg/local_object_storage/metabase/v2/put.go b/pkg/local_object_storage/metabase/v2/put.go
index 8718b47c0..ce6a1e7d3 100644
--- a/pkg/local_object_storage/metabase/v2/put.go
+++ b/pkg/local_object_storage/metabase/v2/put.go
@@ -23,65 +23,67 @@ var ErrUnknownObjectType = errors.New("unknown object type")
 // Put saves object header in metabase. Object payload expected to be cut.
 // Big objects have nil blobovniczaID.
 func (db *DB) Put(obj *object.Object, id *blobovnicza.ID) error {
-	var isParent bool // true when object header obtained from `split.Parent`
+	return db.boltDB.Update(func(tx *bbolt.Tx) error {
+		return db.put(tx, obj, id, false)
+	})
+}
 
-	for ; obj != nil; obj, isParent = obj.GetParent(), true {
-		exists, err := db.Exists(obj.Address())
+func (db *DB) put(tx *bbolt.Tx, obj *object.Object, id *blobovnicza.ID, isParent bool) error {
+	exists, err := db.exists(tx, obj.Address())
+	if err != nil {
+		return err
+	}
+
+	// most right child and split header overlap parent so we have to
+	// check if object exists to not overwrite it twice
+	if exists {
+		return nil
+	}
+
+	if obj.GetParent() != nil && !isParent { // limit depth by two
+		err = db.put(tx, obj.GetParent(), id, true)
 		if err != nil {
 			return err
 		}
+	}
 
-		// most right child and split header overlap parent so we have to
-		// check if object exists to not overwrite it twice
-		if exists {
-			continue
-		}
+	uniqueIndexes, err := uniqueIndexes(obj, isParent, id)
+	if err != nil {
+		return fmt.Errorf("can' build unique indexes: %w", err)
+	}
 
-		uniqueIndexes, err := uniqueIndexes(obj, isParent, id)
+	// put unique indexes
+	for i := range uniqueIndexes {
+		err := putUniqueIndexItem(tx, uniqueIndexes[i])
 		if err != nil {
-			return fmt.Errorf("can' build unique indexes: %w", err)
+			return err
 		}
+	}
 
-		// build list indexes
-		listIndexes, err := listIndexes(obj)
+	// build list indexes
+	listIndexes, err := listIndexes(obj)
+	if err != nil {
+		return fmt.Errorf("can' build list indexes: %w", err)
+	}
+
+	// put list indexes
+	for i := range listIndexes {
+		err := putListIndexItem(tx, listIndexes[i])
 		if err != nil {
-			return fmt.Errorf("can' build list indexes: %w", err)
+			return err
 		}
+	}
 
-		fkbtIndexes, err := fkbtIndexes(obj)
+	// build fake bucket tree indexes
+	fkbtIndexes, err := fkbtIndexes(obj)
+	if err != nil {
+		return fmt.Errorf("can' build fake bucket tree indexes: %w", err)
+	}
+
+	// put fake bucket tree indexes
+	for i := range fkbtIndexes {
+		err := putFKBTIndexItem(tx, fkbtIndexes[i])
 		if err != nil {
-			return fmt.Errorf("can' build fake bucket tree indexes: %w", err)
-		}
-
-		// consider making one TX for both target object and parent
-		err = db.boltDB.Update(func(tx *bbolt.Tx) error {
-			// put unique indexes
-			for i := range uniqueIndexes {
-				err := putUniqueIndexItem(tx, uniqueIndexes[i])
-				if err != nil {
-					return err
-				}
-			}
-
-			// put list indexes
-			for i := range listIndexes {
-				err := putListIndexItem(tx, listIndexes[i])
-				if err != nil {
-					return err
-				}
-			}
-
-			// put fake bucket tree indexes
-			for i := range fkbtIndexes {
-				err := putFKBTIndexItem(tx, fkbtIndexes[i])
-				if err != nil {
-					return err
-				}
-			}
-
-			return nil
-		})
-		if err != nil { // if tx failed then return error
 			return err
 		}
 	}