Add/case 2 and case 4

.github/workflows/tests.yml

@@ -6,10 +6,7 @@ concurrency:
 
 on:
   push:
-    branches:
-      - master
-      - staging
-    tags:
+    branches: [ master, staging, sprint* ]
   pull_request:
 
 jobs:

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

@@ -375,9 +375,7 @@ func TestBlobberCore_CopyFile(t *testing.T) {
 
 			config.Configuration.MaxAllocationDirFiles = tc.maxDirFilesPerAlloc
 
-			ctx := context.TODO()
-			db := datastore.GetStore().GetDB().Begin()
-			ctx = context.WithValue(ctx, datastore.ContextKeyTransaction, db)
+			ctx := datastore.GetStore().CreateTransaction(context.TODO())
 
 			change := &CopyFileChange{
 				AllocationID: tc.allocationID,

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

@@ -2,6 +2,7 @@ package allocation
 
 import (
 	"context"
+
 	"github.com/0chain/blobber/code/go/0chain.net/blobbercore/datastore"
 	"github.com/0chain/blobber/code/go/0chain.net/core/common"
 	"github.com/0chain/errors"
@@ -12,33 +13,25 @@ import (
 // GetOrCreate, get allocation if it exists in db. if not, try to sync it from blockchain, and insert it in db.
 func GetOrCreate(ctx context.Context, store datastore.Store, allocationId string) (*Allocation, error) {
 
-	db := store.GetDB()
+	db := store.CreateTransaction(ctx)
 
 	if len(allocationId) == 0 {
 		return nil, errors.Throw(constants.ErrInvalidParameter, "tx")
 	}
 
-	alloc := &Allocation{}
-	result := db.Table(TableNameAllocation).Where(SQLWhereGetById, allocationId).First(alloc)
-
-	if result.Error == nil {
+	alloc, err := Repo.GetById(db, allocationId)
+	if err == nil {
 		return alloc, nil
 	}
+	if !errors.Is(err, gorm.ErrRecordNotFound) {
 
-	if !errors.Is(result.Error, gorm.ErrRecordNotFound) {
-
-		return nil, errors.ThrowLog(result.Error.Error(), common.ErrBadDataStore)
+		return nil, errors.ThrowLog(err.Error(), common.ErrBadDataStore)
 	}
 
 	return SyncAllocation(allocationId)
 
 }
 
-const (
-	SQLWhereGetByTx = "allocations.tx = ?"
-	SQLWhereGetById = "allocations.id = ?"
-)
-
 // DryRun  Creates a prepared statement when executing any SQL and caches them to speed up future calls
 // https://gorm.io/docs/performance.html#Caches-Prepared-Statement
 func DryRun(db *gorm.DB) {

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

@@ -254,9 +254,7 @@ func TestBlobberCore_DeleteFile(t *testing.T) {
 
 			config.Configuration.MaxAllocationDirFiles = tc.maxDirFilesPerAlloc
 
-			ctx := context.TODO()
-			db := datastore.GetStore().GetDB().Begin()
-			ctx = context.WithValue(ctx, datastore.ContextKeyTransaction, db)
+			ctx := datastore.GetStore().CreateTransaction(context.TODO())
 
 			change := &DeleteFileChange{
 				AllocationID: tc.allocationID,

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

@@ -1,10 +1,12 @@
 package allocation
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"time"
 
+	"github.com/0chain/blobber/code/go/0chain.net/blobbercore/datastore"
 	"github.com/0chain/blobber/code/go/0chain.net/core/common"
 	"gorm.io/gorm/clause"
 
@@ -162,7 +164,8 @@ func (*Pending) TableName() string {
 }
 
 // GetPendingWrite Get write size that is not yet redeemed
-func GetPendingWrite(db *gorm.DB, clientID, allocationID string) (pendingWriteSize int64, err error) {
+func GetPendingWrite(ctx context.Context, clientID, allocationID string) (pendingWriteSize int64, err error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	err = db.Model(&Pending{}).Select("pending_write").Where(
 		"id=?", fmt.Sprintf("%v:%v", clientID, allocationID),
 	).Scan(&pendingWriteSize).Error
@@ -177,7 +180,8 @@ func GetPendingWrite(db *gorm.DB, clientID, allocationID string) (pendingWriteSi
 }
 
 // GetPendingRead Get read size that is not yet redeemed
-func GetPendingRead(db *gorm.DB, clientID, allocationID string) (pendingReadSize int64, err error) {
+func GetPendingRead(ctx context.Context, clientID, allocationID string) (pendingReadSize int64, err error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	err = db.Model(&Pending{}).Select("pending_read").Where(
 		"id=?", fmt.Sprintf("%v:%v", clientID, allocationID),
 	).Scan(&pendingReadSize).Error
@@ -191,7 +195,8 @@ func GetPendingRead(db *gorm.DB, clientID, allocationID string) (pendingReadSize
 	return
 }
 
-func AddToPending(db *gorm.DB, clientID, allocationID string, pendingWrite int64) (err error) {
+func AddToPending(ctx context.Context, clientID, allocationID string, pendingWrite int64) (err error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	key := clientID + ":" + allocationID
 	// Lock is required because two process can simultaneously call this function and read pending data
 	// thus giving same value leading to inconsistent data
@@ -215,7 +220,8 @@ func AddToPending(db *gorm.DB, clientID, allocationID string, pendingWrite int64
 	return nil
 }
 
-func GetWritePoolsBalance(db *gorm.DB, allocationID string) (balance uint64, err error) {
+func GetWritePoolsBalance(ctx context.Context, allocationID string) (balance uint64, err error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	err = db.Model(&WritePool{}).Select("COALESCE(SUM(balance),0) as tot_balance").Where(
 		"allocation_id = ?", allocationID,
 	).Scan(&balance).Error
@@ -266,36 +272,39 @@ func (WritePool) TableName() string {
 	return "write_pools"
 }
 
-func GetReadPool(db *gorm.DB, clientID string) (*ReadPool, error) {
+func GetReadPool(ctx context.Context, clientID string) (*ReadPool, error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	var rp ReadPool
 	return &rp, db.Model(&ReadPool{}).Where("client_id = ?", clientID).Scan(&rp).Error
 }
 
-func GetReadPoolsBalance(db *gorm.DB, clientID string) (int64, error) {
-	rp, err := GetReadPool(db, clientID)
+func GetReadPoolsBalance(ctx context.Context, clientID string) (int64, error) {
+	rp, err := GetReadPool(ctx, clientID)
 	if err != nil {
 		return 0, err
 	}
 
 	return rp.Balance, nil
 }
 
-func UpsertReadPool(db *gorm.DB, rp *ReadPool) error {
+func UpsertReadPool(ctx context.Context, rp *ReadPool) error {
 	updateFields := []string{"balance"}
-
+	db := datastore.GetStore().GetTransaction(ctx)
 	return db.Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "client_id"}},
 		DoUpdates: clause.AssignmentColumns(updateFields), // column needed to be updated
 	}).Create(&rp).Error
 }
 
-func UpdateReadPool(db *gorm.DB, rp *ReadPool) error {
+func UpdateReadPool(ctx context.Context, rp *ReadPool) error {
+	db := datastore.GetStore().GetTransaction(ctx)
 	return db.Model(&ReadPool{}).Where("client_id = ?", rp.ClientID).Updates(map[string]interface{}{
 		"balance": rp.Balance,
 	}).Error
 }
 
-func SetWritePool(db *gorm.DB, allocationID string, wp *WritePool) (err error) {
+func SetWritePool(ctx context.Context, allocationID string, wp *WritePool) (err error) {
+	db := datastore.GetStore().GetTransaction(ctx)
 	err = db.Delete(&WritePool{}, "allocation_id = ?", allocationID).Error
 	if err != nil {
 		return

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

@@ -95,9 +95,7 @@ func TestBlobberCore_FileChangerUpload(t *testing.T) {
 
 			config.Configuration.MaxAllocationDirFiles = tc.maxDirFilesPerAlloc
 
-			ctx := context.TODO()
-			db := datastore.GetStore().GetDB().Begin()
-			ctx = context.WithValue(ctx, datastore.ContextKeyTransaction, db)
+			ctx := datastore.GetStore().CreateTransaction(context.TODO())
 
 			fPath := "/new"
 			change := &UploadFileChanger{

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

@@ -468,9 +468,7 @@ func TestBlobberCore_MoveFile(t *testing.T) {
 
 			config.Configuration.MaxAllocationDirFiles = tc.maxDirFilesPerAlloc
 
-			ctx := context.TODO()
-			db := datastore.GetStore().GetDB().Begin()
-			ctx = context.WithValue(ctx, datastore.ContextKeyTransaction, db)
+			ctx := datastore.GetStore().CreateTransaction(context.TODO())
 
 			change := &MoveFileChange{
 				AllocationID: tc.allocationID,

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

@@ -32,9 +32,7 @@ func TestMultiOp(t *testing.T) {
 	alloc.OwnerPublicKey = sch.GetPublicKey()
 	alloc.OwnerID = client.GetClientID()
 	datastore.MocketTheStore(t, true)
-	ctx := context.TODO()
-	db := datastore.GetStore().GetDB().Begin()
-	ctx = context.WithValue(ctx, datastore.ContextKeyTransaction, db)
+	ctx := datastore.GetStore().CreateTransaction(context.TODO())
 	setupDbMock()
 	fileIDMeta := make(map[string]string)
 	fileIDMeta["/"] = randName()

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

@@ -14,19 +14,12 @@ import (
 	"github.com/0chain/blobber/code/go/0chain.net/core/logging"
 	"github.com/0chain/blobber/code/go/0chain.net/core/node"
 	"github.com/0chain/blobber/code/go/0chain.net/core/transaction"
-	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
 
 // GetAllocationByID from DB. This function doesn't load related terms.
 func GetAllocationByID(ctx context.Context, allocID string) (a *Allocation, err error) {
-	var tx = datastore.GetStore().GetTransaction(ctx)
-
-	a = new(Allocation)
-	err = tx.Model(&Allocation{}).
-		Where("id=?", allocID).
-		First(a).Error
-	return
+	return Repo.GetById(ctx, allocID)
 }
 
 // LoadTerms loads corresponding terms from DB. Since, the GetAllocationByID
@@ -49,14 +42,11 @@ func (a *Allocation) LoadTerms(ctx context.Context) (err error) {
 	return          // found in DB
 }
 
-// VerifyAllocationTransaction try to get allocation from postgres.if it doesn't exists, get it from sharders, and insert it into postgres.
+// FetchAllocationFromEventsDB try to get allocation from postgres.if it doesn't exists, get it from sharders, and insert it into postgres.
 func FetchAllocationFromEventsDB(ctx context.Context, allocationID string, allocationTx string, readonly bool) (a *Allocation, err error) {
 	var tx = datastore.GetStore().GetTransaction(ctx)
 
-	a = new(Allocation)
-	err = tx.Model(&Allocation{}).
-		Where(&Allocation{Tx: allocationTx}).
-		First(a).Error
+	a, err = Repo.GetByTx(ctx, allocationID, allocationTx)
 
 	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil, common.NewError("bad_db_operation", err.Error()) // unexpected DB error
@@ -81,21 +71,13 @@ func FetchAllocationFromEventsDB(ctx context.Context, allocationID string, alloc
 	}
 
 	var isExist bool
-	err = tx.Model(&Allocation{}).
-		Where("id = ?", sa.ID).
-		First(a).Error
+	a, err = Repo.GetById(ctx, sa.ID)
 	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil, common.NewError("bad_db_operation", err.Error()) // unexpected
 	}
 
 	isExist = a.ID != ""
 
-	logging.Logger.Info("VerifyAllocationTransaction",
-		zap.Bool("isExist", isExist),
-		zap.Any("allocation", a),
-		zap.Any("storageAllocation", sa),
-		zap.String("node.Self.ID", node.Self.ID))
-
 	if !isExist {
 		foundBlobber := false
 		for _, blobberConnection := range sa.BlobberDetails {
@@ -136,6 +118,7 @@ func FetchAllocationFromEventsDB(ctx context.Context, allocationID string, alloc
 	if err != nil {
 		return nil, common.NewError("meta_data_update_error", err.Error())
 	}
+
 	// go update allocation data in file store map
 	// related terms
 	a.Terms = make([]*Terms, 0, len(sa.BlobberDetails))
@@ -154,7 +137,8 @@ func FetchAllocationFromEventsDB(ctx context.Context, allocationID string, alloc
 
 	logging.Logger.Info("Saving the allocation to DB")
 
-	err = tx.Save(a).Error
+	err = Repo.Save(ctx, a)
+
 	if err != nil {
 		return nil, err
 	}
@@ -181,13 +165,16 @@ func RequestReadPoolStat(clientID string) (*ReadPool, error) {
 		return nil, fmt.Errorf("requesting read pools stat: %v", err)
 	}
 
-	var readPool ReadPool
-	if err = json.Unmarshal(resp, &readPool); err != nil {
-		return nil, fmt.Errorf("decoding read pools stat response: %v, \n==> resp: %s", err, string(resp))
+	if resp != nil {
+		var readPool ReadPool
+		if err = json.Unmarshal(resp, &readPool); err != nil {
+			return nil, fmt.Errorf("decoding read pools stat response: %v, \n==> resp: %s", err, string(resp))
+		}
+		readPool.ClientID = clientID
+		return &readPool, nil
 	}
 
-	readPool.ClientID = clientID
-	return &readPool, nil
+	return nil, errors.New("empty response received from MakeSCRestAPICall")
 }
 
 func RequestWritePool(allocationID string) (wps *WritePool, err error) {
@@ -205,16 +192,20 @@ func RequestWritePool(allocationID string) (wps *WritePool, err error) {
 		return nil, fmt.Errorf("requesting write pools stat: %v", err)
 	}
 
-	var allocation = struct {
-		ID        string `json:"id"`
-		WritePool uint64 `json:"write_pool"`
-	}{}
-	if err = json.Unmarshal(resp, &allocation); err != nil {
-		return nil, fmt.Errorf("decoding write pools stat response: %v", err)
+	if resp != nil {
+		var allocation = struct {
+			ID        string `json:"id"`
+			WritePool uint64 `json:"write_pool"`
+		}{}
+		if err = json.Unmarshal(resp, &allocation); err != nil {
+			return nil, fmt.Errorf("decoding write pools stat response: %v", err)
+		}
+
+		return &WritePool{
+			AllocationID: allocationID,
+			Balance:      allocation.WritePool,
+		}, nil
 	}
 
-	return &WritePool{
-		AllocationID: allocationID,
-		Balance:      allocation.WritePool,
-	}, nil
+	return nil, errors.New("empty response received from MakeSCRestAPICall")
 }

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

@@ -24,7 +24,7 @@ func (rf *RenameFileChange) DeleteTempFile() error {
 	return nil
 }
 
-func (rf *RenameFileChange) ApplyChange(ctx context.Context, rootRef *reference.Ref, change *AllocationChange,
+func (rf *RenameFileChange) applyChange(ctx context.Context, rootRef *reference.Ref, change *AllocationChange,
 	allocationRoot string, ts common.Timestamp, _ map[string]string) (*reference.Ref, error) {
 
 	if rf.Path == "/" {

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

@@ -0,0 +1,28 @@
+//go:build integration_tests
+// +build integration_tests
+
+package allocation
+
+import (
+	"context"
+	"errors"
+
+	"github.com/0chain/blobber/code/go/0chain.net/blobbercore/reference"
+	"github.com/0chain/blobber/code/go/0chain.net/conductor/conductrpc"
+	"github.com/0chain/blobber/code/go/0chain.net/core/common"
+	"github.com/0chain/blobber/code/go/0chain.net/core/node"
+)
+
+func (rf *RenameFileChange) ApplyChange(ctx context.Context, rootRef *reference.Ref, change *AllocationChange,
+	allocationRoot string, ts common.Timestamp, _ map[string]string) (*reference.Ref, error) {
+
+	state := conductrpc.Client().State()
+	if state.FailRenameCommit != nil {
+		for _, nodeId := range state.FailRenameCommit {
+			if nodeId == node.Self.ID {
+				return nil, errors.New("error directed by conductor")
+			}
+		}
+	}
+	return rf.applyChange(ctx, rootRef, change, allocationRoot, ts, nil)
+}