add check for chunkNumber (#1253)

* add check for chunkNumber * add err log * fix save progress
0chain · Oct 11, 2023 · 4c07b1f · 4c07b1f
1 parent 3d9659e
commit 4c07b1f
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 11 deletions.
diff --git a/zboxcore/sdk/chunked_upload.go b/zboxcore/sdk/chunked_upload.go
@@ -218,7 +218,7 @@ func CreateChunkedUpload(
 	// encrypt option has been changed. upload it from scratch
 	// chunkSize has been changed. upload it from scratch
 	// actual size has been changed. upload it from scratch
-	if su.progress.ChunkSize != su.chunkSize || su.progress.EncryptOnUpload != su.encryptOnUpload || su.progress.ActualSize != su.fileMeta.ActualSize {
+	if su.progress.ChunkSize != su.chunkSize || su.progress.EncryptOnUpload != su.encryptOnUpload || su.progress.ActualSize != su.fileMeta.ActualSize || su.progress.ChunkNumber != su.chunkNumber || su.progress.ConnectionID == "" {
 		su.progress.ChunkSize = 0 // reset chunk size
 	}
 
@@ -336,6 +336,7 @@ func (su *ChunkedUpload) createUploadProgress(connectionId string) {
 			EncryptOnUpload:   su.encryptOnUpload,
 			EncryptedKeyPoint: su.encryptedKeyPoint,
 			ActualSize:        su.fileMeta.ActualSize,
+			ChunkNumber:       su.chunkNumber,
 		}
 	}
 	su.progress.Blobbers = make([]*UploadBlobberStatus, su.allocationObj.DataShards+su.allocationObj.ParityShards)
@@ -439,15 +440,16 @@ func (su *ChunkedUpload) process() error {
 		} else {
 			// Write data to hashers
 			for i, blobberShard := range chunks.fileShards {
+				hasher := su.blobbers[i].progress.Hasher
 				for _, chunkBytes := range blobberShard {
-					err = su.blobbers[i].progress.Hasher.WriteToFixedMT(chunkBytes)
+					err = hasher.WriteToFixedMT(chunkBytes)
 					if err != nil {
 						if su.statusCallback != nil {
 							su.statusCallback.Error(su.allocationObj.ID, su.fileMeta.RemotePath, su.opCode, err)
 						}
 						return err
 					}
-					err = su.blobbers[i].progress.Hasher.WriteToValidationMT(chunkBytes)
+					err = hasher.WriteToValidationMT(chunkBytes)
 					if err != nil {
 						if su.statusCallback != nil {
 							su.statusCallback.Error(su.allocationObj.ID, su.fileMeta.RemotePath, su.opCode, err)
@@ -462,9 +464,6 @@ func (su *ChunkedUpload) process() error {
 		// last chunk might 0 with io.EOF
 		// https://stackoverflow.com/questions/41208359/how-to-test-eof-on-io-reader-in-go
 		if chunks.totalReadSize > 0 && chunks.chunkEndIndex > su.progress.ChunkIndex {
-			su.progress.ChunkIndex = chunks.chunkEndIndex
-			su.saveProgress()
-
 			if su.statusCallback != nil {
 				su.statusCallback.InProgress(su.allocationObj.ID, su.fileMeta.RemotePath, su.opCode, int(su.progress.UploadLength)-alreadyUploadedData, nil)
 			}
@@ -590,6 +589,7 @@ func (su *ChunkedUpload) processUpload(chunkStartIndex, chunkEndIndex int,
 		isFinal:         isFinal,
 		encryptedKey:    su.encryptedKey,
 		uploadBody:      make([]blobberData, len(su.blobbers)),
+		saveProgress:    uploadLength > 0,
 	}
 
 	wgErrors := make(chan error, len(su.blobbers))
@@ -784,6 +784,10 @@ func (su *ChunkedUpload) uploadProcessor() {
 					su.fileMeta.RemotePath, su.consensus.consensusThresh, su.consensus.getConsensus())))
 				return
 			}
+			if uploadData.saveProgress {
+				su.progress.ChunkIndex = uploadData.chunkEndIndex
+				su.saveProgress()
+			}
 			if uploadData.isFinal {
 				return
 			}

diff --git a/zboxcore/sdk/chunked_upload_model.go b/zboxcore/sdk/chunked_upload_model.go
@@ -169,8 +169,9 @@ type UploadProgress struct {
 	ID string `json:"id"`
 
 	// ChunkSize size of chunk
-	ChunkSize  int64 `json:"chunk_size,omitempty"`
-	ActualSize int64 `json:"actual_size,omitempty"`
+	ChunkSize   int64 `json:"chunk_size,omitempty"`
+	ActualSize  int64 `json:"actual_size,omitempty"`
+	ChunkNumber int   `json:"chunk_number,omitempty"`
 	// EncryptOnUpload encrypt data on upload or not
 	EncryptOnUpload   bool   `json:"is_encrypted,omitempty"`
 	EncryptPrivateKey string `json:"-"`
@@ -200,6 +201,7 @@ type UploadData struct {
 	chunkStartIndex int
 	chunkEndIndex   int
 	isFinal         bool
+	saveProgress    bool
 	encryptedKey    string
 	uploadBody      []blobberData
 }

diff --git a/zboxcore/sdk/multi_operation_worker.go b/zboxcore/sdk/multi_operation_worker.go
@@ -297,6 +297,7 @@ func (mo *MultiOperation) Process() error {
 			fmt.Sprintf("Commit failed. Required consensus %d, got %d",
 				mo.Consensus.consensusThresh, mo.Consensus.consensus))
 		if mo.getConsensus() != 0 {
+			l.Logger.Info("Rolling back changes on minority blobbers")
 			mo.allocationObj.RollbackWithMask(rollbackMask)
 		}
 		for _, op := range mo.operations {

diff --git a/zboxcore/zboxutil/http.go b/zboxcore/zboxutil/http.go
@@ -22,8 +22,6 @@ import (
 	"github.com/0chain/gosdk/core/logger"
 	"github.com/0chain/gosdk/zboxcore/blockchain"
 	"github.com/0chain/gosdk/zboxcore/client"
-
-	l "github.com/0chain/gosdk/zboxcore/logger"
 )
 
 const SC_REST_API_URL = "v1/screst/"
@@ -576,7 +574,6 @@ func NewConnectionRequest(baseUrl, allocationID string, allocationTx string, bod
 	if err != nil {
 		return nil, err
 	}
-	l.Logger.Error("request url: ", u)
 	req, err := http.NewRequest(http.MethodPost, u.String(), body)
 	if err != nil {
 		return nil, err