Fix cleanup worker (#1262)

* add logs to check alloc id

* fix empty allocation id

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

@@ -195,7 +195,6 @@ func (cc *AllocationChangeCollector) ComputeProperties() {
 		if acp == nil {
 			continue // unknown operation (impossible case?)
 		}
-
 		if err := acp.Unmarshal(change.Input); err != nil { // error is not handled
 			logging.Logger.Error("AllocationChangeCollector_unmarshal", zap.Error(err))
 		}

code/go/0chain.net/blobbercore/challenge/challenge.go

@@ -107,25 +107,19 @@ func syncOpenChallenges(ctx context.Context) {
 
 }
 
-func validateOnValidators(c *ChallengeEntity) {
+func validateOnValidators(ctx context.Context, c *ChallengeEntity) error {
 
 	logging.Logger.Info("[challenge]validate: ",
 		zap.Any("challenge", c),
 		zap.String("challenge_id", c.ChallengeID),
 	)
 
-	ctx := datastore.GetStore().CreateTransaction(context.TODO())
-	defer ctx.Done()
-
-	tx := datastore.GetStore().GetTransaction(ctx)
-
 	if err := CreateChallengeTiming(c.ChallengeID, c.CreatedAt); err != nil {
 		logging.Logger.Error("[challengetiming]add: ",
 			zap.String("challenge_id", c.ChallengeID),
 			zap.Error(err))
 		deleteChallenge(c.RoundCreatedAt)
-		tx.Rollback()
-		return
+		return err
 	}
 
 	createdTime := common.ToTime(c.CreatedAt)
@@ -143,10 +137,8 @@ func validateOnValidators(c *ChallengeEntity) {
 			zap.String("validationTickets", string(c.ValidationTicketsString)),
 			zap.String("ObjectPath", string(c.ObjectPathString)),
 			zap.Error(err))
-		tx.Rollback()
-
 		c.CancelChallenge(ctx, err)
-		return
+		return nil
 	}
 
 	if err := c.LoadValidationTickets(ctx); err != nil {
@@ -156,17 +148,7 @@ func validateOnValidators(c *ChallengeEntity) {
 			zap.Error(err))
 		//TODO: Should we delete the challenge from map or send it back to the todo channel?
 		deleteChallenge(c.RoundCreatedAt)
-		tx.Rollback()
-		return
-	}
-
-	if err := tx.Commit().Error; err != nil {
-		logging.Logger.Error("[challenge]validate(Commit): ",
-			zap.Any("challenge_id", c.ChallengeID),
-			zap.Time("created", createdTime),
-			zap.Error(err))
-		tx.Rollback()
-		return
+		return err
 	}
 
 	completedValidation := time.Now()
@@ -181,7 +163,7 @@ func validateOnValidators(c *ChallengeEntity) {
 	logging.Logger.Info("[challenge]validate: ",
 		zap.Any("challenge_id", c.ChallengeID),
 		zap.Time("created", createdTime))
-
+	return nil
 }
 
 func (c *ChallengeEntity) getCommitTransaction() (*transaction.Transaction, error) {

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

@@ -111,7 +111,9 @@ func processChallenge(ctx context.Context, it *ChallengeEntity) {
 	logging.Logger.Info("processing_challenge",
 		zap.String("challenge_id", it.ChallengeID))
 
-	validateOnValidators(it)
+	_ = datastore.GetStore().WithNewTransaction(func(ctx context.Context) error {
+		return validateOnValidators(ctx, it)
+	})
 }
 
 func commitOnChainWorker(ctx context.Context) {

code/go/0chain.net/blobbercore/filestore/storage.go

@@ -371,6 +371,10 @@ func (fs *FileStore) DeleteFile(allocID, validationRoot string) error {
 }
 
 func (fs *FileStore) DeleteTempFile(allocID, conID string, fd *FileInputData) error {
+	if allocID == "" {
+		logging.Logger.Error("invalid_allocation_id", zap.String("connection_id", conID), zap.Any("file_data", fd))
+		return common.NewError("invalid_allocation_id", "Allocation id cannot be empty")
+	}
 	fileObjectPath := fs.getTempPathForFile(allocID, fd.Name, encryption.Hash(fd.Path), conID)
 
 	finfo, err := os.Stat(fileObjectPath)

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

@@ -215,6 +215,7 @@ func (cmd *UpdateFileCommand) reloadChange(connectionObj *allocation.AllocationC
 
 // UpdateChange add UpdateFileChanger in db
 func (cmd *UpdateFileCommand) UpdateChange(ctx context.Context, connectionObj *allocation.AllocationChangeCollector) error {
+	cmd.fileChanger.AllocationID = connectionObj.AllocationID
 	for _, c := range connectionObj.Changes {
 		filePath, _ := c.GetOrParseAffectedFilePath()
 		if c.Operation != sdkConst.FileOperationUpdate || cmd.fileChanger.Path != filePath {

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

@@ -223,6 +223,7 @@ func (cmd *UploadFileCommand) reloadChange(connectionObj *allocation.AllocationC
 
 // UpdateChange replace AddFileChange in db
 func (cmd *UploadFileCommand) UpdateChange(ctx context.Context, connectionObj *allocation.AllocationChangeCollector) error {
+	cmd.fileChanger.AllocationID = connectionObj.AllocationID
 	for _, c := range connectionObj.Changes {
 		filePath, _ := c.GetOrParseAffectedFilePath()
 		if c.Operation != constants.FileOperationInsert || cmd.fileChanger.Path != filePath {

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

@@ -224,7 +224,7 @@ func setupHandlers(r *mux.Router) {
 	// r.HandleFunc("/_statsJSON", common.AuthenticateAdmin(common.ToJSONResponse(stats.StatsJSONHandler)))
 	r.HandleFunc("/_statsJSON", RateLimitByCommmitRL(common.ToJSONResponse(stats.StatsJSONHandler)))
 	// r.HandleFunc("/_cleanupdisk", common.AuthenticateAdmin(common.ToJSONResponse(WithReadOnlyConnection(CleanupDiskHandler))))
-	r.HandleFunc("/_cleanupdisk", RateLimitByCommmitRL(common.ToJSONResponse(WithReadOnlyConnection(CleanupDiskHandler))))
+	// r.HandleFunc("/_cleanupdisk", RateLimitByCommmitRL(common.ToJSONResponse(WithReadOnlyConnection(CleanupDiskHandler))))
 	// r.HandleFunc("/getstats", common.AuthenticateAdmin(common.ToJSONResponse(stats.GetStatsHandler)))
 	r.HandleFunc("/getstats", RateLimitByCommmitRL(common.ToJSONResponse(WithReadOnlyConnection(stats.GetStatsHandler))))
 	// r.HandleFunc("/challengetimings", common.AuthenticateAdmin(common.ToJSONResponse(GetChallengeTimings)))
@@ -271,11 +271,12 @@ func setupHandlers(r *mux.Router) {
 func WithReadOnlyConnection(handler common.JSONResponderF) common.JSONResponderF {
 	return func(ctx context.Context, r *http.Request) (interface{}, error) {
 		ctx = GetMetaDataStore().CreateTransaction(ctx)
-
-		res, err := handler(ctx, r)
+		tx := GetMetaDataStore().GetTransaction(ctx)
 		defer func() {
-			GetMetaDataStore().GetTransaction(ctx).Rollback()
+			tx.Rollback()
 		}()
+
+		res, err := handler(ctx, r)
 		return res, err
 	}
 }