Fix phase locking (#1230)

* add logs for 2 phase lock

* add update log

* add root in log

* add update log

* use update lock

* add log in update repo

* use save

* use exec

* log rows affected

* use no key update

* use repo update

* cleanup

* add defer

* add defer

Author: Hitenjain14

code/go/0chain.net/blobbercore/allocation/repository.go

@@ -47,7 +47,7 @@ func (r *Repository) GetById(ctx context.Context, id string) (*Allocation, error
 	}
 
 	alloc := &Allocation{}
-	err = tx.Table(TableNameAllocation).Where(SQLWhereGetById, id).First(alloc).Error
+	err = tx.Table(TableNameAllocation).Where(SQLWhereGetById, id).Take(alloc).Error
 	if err != nil {
 		return alloc, err
 	}
@@ -73,7 +73,7 @@ func (r *Repository) GetByIdAndLock(ctx context.Context, id string) (*Allocation
 	err = tx.Model(&Allocation{}).
 		Clauses(clause.Locking{Strength: "NO KEY UPDATE"}).
 		Where("id=?", id).
-		First(alloc).Error
+		Take(alloc).Error
 	if err != nil {
 		return alloc, err
 	}
@@ -99,7 +99,7 @@ func (r *Repository) GetByTx(ctx context.Context, allocationID, txHash string) (
 	}
 
 	alloc := &Allocation{}
-	err = tx.Table(TableNameAllocation).Where(SQLWhereGetByTx, txHash).First(alloc).Error
+	err = tx.Table(TableNameAllocation).Where(SQLWhereGetByTx, txHash).Take(alloc).Error
 	if err != nil {
 		return alloc, err
 	}
@@ -139,7 +139,7 @@ func (r *Repository) GetAllocationIds(ctx context.Context) []Res {
 
 }
 
-func (r *Repository) UpdateAllocationRedeem(ctx context.Context, allocationID, AllocationRoot string) error {
+func (r *Repository) UpdateAllocationRedeem(ctx context.Context, allocationID, AllocationRoot string, allocationObj *Allocation) error {
 	var tx = datastore.GetStore().GetTransaction(ctx)
 	if tx == nil {
 		logging.Logger.Panic("no transaction in the context")
@@ -151,9 +151,10 @@ func (r *Repository) UpdateAllocationRedeem(ctx context.Context, allocationID, A
 	}
 	delete(cache, allocationID)
 
-	err = tx.Exec("UPDATE allocations SET latest_redeemed_write_marker=?,is_redeem_required=? WHERE id=?",
-		AllocationRoot, false, allocationID).Error
-
+	allocationUpdates := make(map[string]interface{})
+	allocationUpdates["latest_redeemed_write_marker"] = AllocationRoot
+	allocationUpdates["is_redeem_required"] = false
+	err = tx.Model(allocationObj).Updates(allocationUpdates).Error
 	return err
 }
 

code/go/0chain.net/blobbercore/handler/object_operation_handler.go

@@ -480,7 +480,6 @@ func (fsh *StorageHandler) CommitWrite(ctx context.Context, r *http.Request) (*b
 	if err != nil {
 		return nil, common.NewError("invalid_parameters", "Invalid allocation id passed."+err.Error())
 	}
-
 	if allocationObj.FileOptions == 0 {
 		return nil, common.NewError("immutable_allocation", "Cannot write to an immutable allocation")
 	}
@@ -1407,6 +1406,7 @@ func (fsh *StorageHandler) Rollback(ctx context.Context, r *http.Request) (*blob
 	Logger.Info("rollback_writemarker", zap.Any("writemarker", writemarkerEntity.WM))
 
 	alloc, err := allocation.Repo.GetByIdAndLock(c, allocationID)
+	Logger.Info("[rollback]Lock Allocation", zap.Bool("is_redeem_required", alloc.IsRedeemRequired), zap.String("allocation_root", alloc.AllocationRoot), zap.String("latest_wm_redeemed", alloc.LatestRedeemedWM))
 	if err != nil {
 		txn.Rollback()
 		return &result, common.NewError("allocation_read_error", "Error reading the allocation object")

code/go/0chain.net/blobbercore/writemarker/worker.go

@@ -71,7 +71,9 @@ func redeemWriteMarker(wm *WriteMarkerEntity) error {
 	alloc, err := allocation.Repo.GetByIdAndLock(ctx, allocationID)
 	if err != nil {
 		logging.Logger.Error("Error redeeming the write marker.", zap.Any("allocation", allocationID), zap.Any("wm", wm.WM.AllocationID), zap.Any("error", err))
-		go tryAgain(wm)
+		if err != gorm.ErrRecordNotFound {
+			go tryAgain(wm)
+		}
 		shouldRollback = true
 		return err
 	}
@@ -80,12 +82,12 @@ func redeemWriteMarker(wm *WriteMarkerEntity) error {
 		logging.Logger.Info("Stale write marker. Allocation root mismatch",
 			zap.Any("allocation", allocationID),
 			zap.Any("wm", wm.WM.AllocationRoot), zap.Any("alloc_root", alloc.AllocationRoot))
+		_ = wm.UpdateStatus(ctx, Rollbacked, "rollbacked", "")
+		err = db.Commit().Error
 		mut := GetLock(allocationID)
 		if mut != nil {
 			mut.Release(1)
 		}
-		_ = wm.UpdateStatus(ctx, Rollbacked, "rollbacked", "")
-		err = db.Commit().Error
 		return err
 	}
 
@@ -100,13 +102,13 @@ func redeemWriteMarker(wm *WriteMarkerEntity) error {
 
 		return err
 	}
-	mut := GetLock(allocationID)
-	if mut != nil {
-		mut.Release(1)
-	}
-
-	err = allocation.Repo.UpdateAllocationRedeem(ctx, wm.WM.AllocationRoot, allocationID)
-
+	defer func() {
+		mut := GetLock(allocationID)
+		if mut != nil {
+			mut.Release(1)
+		}
+	}()
+	err = allocation.Repo.UpdateAllocationRedeem(ctx, wm.WM.AllocationRoot, allocationID, alloc)
 	if err != nil {
 		logging.Logger.Error("Error redeeming the write marker. Allocation latest wm redeemed update failed",
 			zap.Any("allocation", allocationID),