Merge branch 'sprint-1.11' into feat/l2cache

.github/workflows/build-&-publish-docker-image.yml

@@ -26,7 +26,7 @@ jobs:
     steps:
       - name: Cleanup before restarting conductor tests.
         run: |
-          echo 'y' | docker system prune -a
+          echo 'y' | docker system prune -a || true
           cd /tmp
           sudo rm -rf ./*
 
@@ -99,7 +99,7 @@ jobs:
     steps:
       - name: Cleanup before restarting conductor tests.
         run: |
-          echo 'y' | docker system prune -a
+          echo 'y' | docker system prune -a  || true
           cd /tmp
           sudo rm -rf ./*
 

code/go/0chain.net/blobber/main.go

@@ -48,8 +48,13 @@ func main() {
 		panic(err)
 	}
 
-	err := setCCTFromChain()
-	if err != nil {
+	if err := setCCTFromChain(); err != nil {
+		logging.Logger.Error("Error setCCTFromChain" + err.Error())
+		panic(err)
+	}
+
+	if err := setMaxFileSizeFromChain(); err != nil {
+		logging.Logger.Error("Error setMaxFileSizeFromChain" + err.Error())
 		panic(err)
 	}
 

code/go/0chain.net/blobber/settings.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"go.uber.org/zap"
 	"strconv"
 	"time"
 
@@ -18,6 +19,12 @@ type cctCB struct {
 	err  error
 }
 
+type maxFileSizeCB struct {
+	done chan struct{}
+	mfs  int64
+	err  error
+}
+
 func (c *cctCB) OnInfoAvailable(op int, status int, info string, errStr string) {
 	defer func() {
 		c.done <- struct{}{}
@@ -48,6 +55,38 @@ func (c *cctCB) OnInfoAvailable(op int, status int, info string, errStr string)
 	c.cct = cct
 }
 
+func (c *maxFileSizeCB) OnInfoAvailable(op int, status int, info string, errStr string) {
+	defer func() {
+		c.done <- struct{}{}
+	}()
+
+	if errStr != "" {
+		c.err = errors.New(errStr)
+		return
+	}
+
+	m := make(map[string]interface{})
+	err := json.Unmarshal([]byte(info), &m)
+	if err != nil {
+		c.err = err
+		return
+	}
+
+	m = m["fields"].(map[string]interface{})
+
+	mfsString := m["max_file_size"].(string)
+
+	mfs, err := strconv.ParseInt(mfsString, 10, 64)
+	if err != nil {
+		c.err = err
+		return
+	}
+
+	logging.Logger.Info("max file size from chain", zap.Int64("max_file_size", mfs))
+
+	c.mfs = mfs
+}
+
 func setCCTFromChain() error {
 	cb := &cctCB{
 		done: make(chan struct{}),
@@ -65,6 +104,25 @@ func setCCTFromChain() error {
 	return nil
 }
 
+func setMaxFileSizeFromChain() error {
+	logging.Logger.Info("getting max file size from chain")
+
+	cb := &maxFileSizeCB{
+		done: make(chan struct{}),
+	}
+	err := zcncore.GetStorageSCConfig(cb)
+	if err != nil {
+		return err
+	}
+	<-cb.done
+	if cb.err != nil {
+		return err
+	}
+
+	config.StorageSCConfig.MaxFileSize = cb.mfs
+	return nil
+}
+
 func updateCCTWorker(ctx context.Context) {
 	ticker := time.NewTicker(time.Hour)
 

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

@@ -50,7 +50,7 @@ var (
 	roundInfo          = RoundInfo{}
 )
 
-const batchSize = 10
+const batchSize = 5
 
 // SetupWorkers start challenge workers
 func SetupWorkers(ctx context.Context) {
@@ -217,23 +217,16 @@ func getBatch(batchSize int) (chall []ChallengeEntity) {
 		return
 	}
 
-	// create a set of string
-	allocations := make(map[string]bool)
-
 	it := challengeMap.Iterator()
 	for it.Next() {
 		if len(chall) >= batchSize {
 			break
 		}
-
 		ticket := it.Value().(*ChallengeEntity)
-
-		_, ok := allocations[ticket.AllocationID]
-		if ticket.Status != Processed || ok {
-			continue
+		if ticket.Status != Processed {
+			break
 		}
 		chall = append(chall, *ticket)
-		allocations[ticket.AllocationID] = true
 	}
 	return
 }

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

@@ -187,6 +187,7 @@ func ValidChain(chain string) error {
 // good to go
 type StorageSCConfiguration struct {
 	ChallengeCompletionTime int64
+	MaxFileSize             int64
 }
 
 var StorageSCConfig StorageSCConfiguration

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/0chain/blobber/code/go/0chain.net/blobbercore/config"
 	"mime/multipart"
 	"net/http"
 
@@ -62,6 +63,11 @@ func (cmd *UpdateFileCommand) IsValidated(ctx context.Context, req *http.Request
 			"Invalid parameters. Error parsing the meta data for upload."+err.Error())
 	}
 
+	if cmd.fileChanger.Size > config.StorageSCConfig.MaxFileSize {
+		return common.NewError("max_file_size",
+			fmt.Sprintf("file size %d should not be greater than %d", cmd.fileChanger.Size, config.StorageSCConfig.MaxFileSize))
+	}
+
 	if cmd.fileChanger.ConnectionID == "" {
 		return common.NewError("invalid_connection", "Invalid connection id")
 	}

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/0chain/blobber/code/go/0chain.net/blobbercore/config"
 	"mime/multipart"
 	"net/http"
 	"path/filepath"
@@ -61,6 +62,11 @@ func (cmd *UploadFileCommand) IsValidated(ctx context.Context, req *http.Request
 			"Invalid parameters. Error parsing the meta data for upload."+err.Error())
 	}
 
+	if fileChanger.Size > config.StorageSCConfig.MaxFileSize {
+		return common.NewError("max_file_size",
+			fmt.Sprintf("file size %d should not be greater than %d", fileChanger.Size, config.StorageSCConfig.MaxFileSize))
+	}
+
 	if fileChanger.Path == "/" {
 		return common.NewError("invalid_path", "Invalid path. Cannot upload to root directory")
 	}

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

@@ -75,6 +75,7 @@ func init() {
 	config.Configuration.SignatureScheme = "bls0chain"
 	blobConfig.Configuration.BlockLimitDaily = 1562500
 	blobConfig.Configuration.BlockLimitRequest = 500
+	blobConfig.StorageSCConfig.MaxFileSize = 1024 * 1024 * 1024 * 1024 * 5
 	logging.Logger = zap.NewNop()
 	ConfigRateLimits()