Merge pull request #1535 from 0chain/feat/clear-precommit

Clean precommit dir

Author: dabasov

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

@@ -449,6 +449,11 @@ func (a *AllocationChangeCollector) MoveToFilestore(ctx context.Context, allocat
 			logging.Logger.Error("Error while moving to filestore", zap.Error(err))
 			return err
 		}
+		err = filestore.GetFileStore().DeletePreCommitDir(a.AllocationID)
+		if err != nil {
+			logging.Logger.Error("Error while deleting precommit dir", zap.Error(err))
+			return err
+		}
 
 		return tx.Exec("UPDATE reference_objects SET is_precommit=?, prev_validation_root=validation_root, prev_thumbnail_hash=thumbnail_hash WHERE allocation_id=? AND is_precommit=? AND deleted_at is NULL", false, a.AllocationID, true).Error
 	})

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

@@ -244,8 +244,13 @@ func DeleteConnectionObjEntry(connectionID string) {
 	connectionObj, ok := connectionProcessor[connectionID]
 	if ok {
 		connectionObj.cnclCtx()
+		for _, change := range connectionObj.changes {
+			if change.seqPQ != nil {
+				change.seqPQ.Done(seqpriorityqueue.UploadData{}, 1)
+			}
+		}
+		delete(connectionProcessor, connectionID)
 	}
-	delete(connectionProcessor, connectionID)
 	connectionObjMutex.Unlock()
 }
 

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

@@ -46,6 +46,7 @@ import (
 	"github.com/0chain/blobber/code/go/0chain.net/core/encryption"
 	"github.com/0chain/blobber/code/go/0chain.net/core/logging"
 	"github.com/0chain/gosdk/core/util"
+	"github.com/remeh/sizedwaitgroup"
 	"go.uber.org/zap"
 	"golang.org/x/crypto/sha3"
 	"golang.org/x/sys/unix"
@@ -201,12 +202,55 @@ func (fs *FileStore) DeleteFromFilestore(allocID, hash string, version int) erro
 }
 
 func (fs *FileStore) DeletePreCommitDir(allocID string) error {
-
+	now := time.Now()
 	preCommitDir := fs.getPreCommitDir(allocID)
-	err := os.RemoveAll(preCommitDir)
+	swg := sizedwaitgroup.New(5)
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+
+	entries, err := os.ReadDir(preCommitDir)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil
+		}
+		return common.NewError("pre_commit_dir_read_error", err.Error())
+	}
+
+	for _, entry := range entries {
+		entryPath := filepath.Join(preCommitDir, entry.Name())
+		if entry.IsDir() {
+			swg.Add()
+			go func() {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					err := os.RemoveAll(entryPath)
+					if err != nil {
+						logging.Logger.Error("failed to remove directory", zap.String("path", entryPath), zap.Error(err))
+						cancel(err)
+					}
+				}
+				swg.Done()
+			}()
+		} else {
+			if err := os.Remove(entryPath); err != nil {
+				return fmt.Errorf("failed to remove file %s: %w", entryPath, err)
+			}
+		}
+	}
+	swg.Wait()
+	select {
+	case <-ctx.Done():
+		return common.NewError("pre_commit_dir_deletion_error", ctx.Err().Error())
+	default:
+	}
+
+	err = os.Remove(preCommitDir)
 	if err != nil {
 		return common.NewError("pre_commit_dir_deletion_error", err.Error())
 	}
+	logging.Logger.Debug("pre_commit_dir_deleted", zap.String("allocation_id", allocID), zap.Duration("elapsed", time.Since(now)))
 	return nil
 }
 

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

@@ -83,6 +83,10 @@ func cleanupTempFiles(ctx context.Context) {
 
 	for i := 0; i < len(openConnectionsToDelete); i++ {
 		connection := &openConnectionsToDelete[i]
+		processor := allocation.GetConnectionProcessor(connection.ID)
+		if processor != nil {
+			continue
+		}
 		logging.Logger.Info("Deleting temp files for the connection", zap.Any("connection", connection.ID))
 		connection.ComputeProperties()