@peter/refactor (#122)

* refactor so sdk looks more like sdk

* refactor test cases

* remove sha3

Author: 0g-peterzhb

.github/workflows/go.yml

@@ -14,12 +14,12 @@ jobs:
   unit-test:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v5
 
     - name: Set up Go
       uses: actions/setup-go@v4
       with:
-        go-version: '1.21.5'
+        go-version: '1.22.5'
 
     - name: Build
       run: go build ./...

.github/workflows/tests.yml

@@ -29,7 +29,7 @@ jobs:
         swap-storage: true
 
     - name: Checkout sources
-      uses: actions/checkout@v4
+      uses: actions/checkout@v5
       with:
         submodules: recursive
 
@@ -49,10 +49,10 @@ jobs:
     - name: Build
       run: go build 
 
-    - name: Set up Python 3.9
+    - name: Set up Python 3.11.9
       uses: actions/setup-python@v4
       with:
-        python-version: '3.9'
+        python-version: '3.11.9'
         cache: 'pip'
 
     - name: Install dependencies

cmd/upload.go

@@ -11,12 +11,9 @@ import (
 	"github.com/0gfoundation/0g-storage-client/common/blockchain"
 	"github.com/0gfoundation/0g-storage-client/core"
 	"github.com/0gfoundation/0g-storage-client/indexer"
-	"github.com/0gfoundation/0g-storage-client/node"
 	"github.com/0gfoundation/0g-storage-client/transfer"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
-	"github.com/openweb3/web3go"
-	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -170,94 +167,55 @@ func upload(*cobra.Command, []string) {
 	}
 	defer file.Close()
 
-	uploader, closer, err := newUploader(ctx, file.NumSegments(), uploadArgs, w3client, opt)
-	if err != nil {
-		logrus.WithError(err).Fatal("Failed to initialize uploader")
-	}
-	defer closer()
-	uploader.WithRoutines(uploadArgs.routines)
-
-	_, roots, err := uploader.SplitableUpload(ctx, file, uploadArgs.fragmentSize, opt)
-	if err != nil && opt.FullTrusted {
-		logrus.WithError(err).Fatal("Failed to upload file with full trusted nodes")
-	}
-
-	if err != nil && !opt.FullTrusted {
-		logrus.WithError(err).Warn("Upload with full trusted nodes failed, retrying with all full trusted nodes")
-		opt.FullTrusted = true
-		fullUploader, fullCloser, err := newUploader(ctx, file.NumSegments(), uploadArgs, w3client, opt)
+	var roots []common.Hash
+	if uploadArgs.indexer != "" {
+		indexerClient, err := indexer.NewClient(uploadArgs.indexer, indexer.IndexerClientOption{
+			ProviderOption: providerOption,
+			LogOption:      zg_common.LogOption{Logger: logrus.StandardLogger()},
+			Routines:       uploadArgs.routines,
+			Contract: &transfer.ContractAddress{
+				FlowAddress:   uploadArgs.flowAddress,
+				MarketAddress: uploadArgs.marketAddress,
+			},
+		})
 		if err != nil {
-			logrus.WithError(err).Fatal("Failed to initialize uploader")
+			logrus.WithError(err).Fatal("Failed to initialize indexer client")
 		}
-		defer fullCloser()
-		fullUploader.WithRoutines(uploadArgs.routines)
-		_, roots, err = fullUploader.SplitableUpload(ctx, file, uploadArgs.fragmentSize, opt)
+		defer indexerClient.Close()
+
+		_, roots, err = indexerClient.SplitableUpload(ctx, w3client, file, uploadArgs.fragmentSize, opt)
 		if err != nil {
-			logrus.WithError(err).Fatal("Failed to upload file with full trusted nodes")
+			logrus.WithError(err).Fatal("Failed to upload file")
 		}
-	}
-
-	if len(roots) == 1 {
-		logrus.Infof("file uploaded, root = %v", roots[0])
 	} else {
-		s := make([]string, len(roots))
-		for i, root := range roots {
-			s[i] = root.String()
-		}
-		logrus.Infof("file uploaded in %v fragments, roots = %v", len(roots), strings.Join(s, ","))
-	}
-}
-
-func newUploader(ctx context.Context, segNum uint64, args uploadArgument, w3client *web3go.Client, opt transfer.UploadOption) (*transfer.Uploader, func(), error) {
-	var contractConfig *transfer.UploaderContractConfig
-	if args.flowAddress != "" {
-		if !common.IsHexAddress(args.flowAddress) {
-			return nil, nil, errors.Errorf("invalid flow address: %s", args.flowAddress)
-		}
-		flowAddr := common.HexToAddress(args.flowAddress)
-		contractConfig = &transfer.UploaderContractConfig{
-			FlowAddress: &flowAddr,
-		}
-		if args.marketAddress != "" {
-			if !common.IsHexAddress(args.marketAddress) {
-				return nil, nil, errors.Errorf("invalid market address: %s", args.marketAddress)
-			}
-			marketAddr := common.HexToAddress(args.marketAddress)
-			contractConfig.MarketAddress = &marketAddr
-		}
-	} else if args.marketAddress != "" {
-		return nil, nil, errors.New("market-address requires flow-address")
-	}
-
-	if args.indexer != "" {
-		indexerClient, err := indexer.NewClient(args.indexer, indexer.IndexerClientOption{
+		uploader, closer, err := transfer.NewUploaderFromConfig(ctx, w3client, transfer.UploaderConfig{
+			Nodes:          uploadArgs.node,
 			ProviderOption: providerOption,
 			LogOption:      zg_common.LogOption{Logger: logrus.StandardLogger()},
+			Contact: &transfer.ContractAddress{
+				FlowAddress:   uploadArgs.flowAddress,
+				MarketAddress: uploadArgs.marketAddress,
+			},
+			Routines: uploadArgs.routines,
 		})
 		if err != nil {
-			return nil, nil, errors.WithMessage(err, "failed to initialize indexer client")
+			logrus.WithError(err).Fatal("Failed to initialize uploader")
 		}
+		defer closer()
 
-		up, err := indexerClient.NewUploaderFromIndexerNodesWithContractConfig(ctx, segNum, w3client, opt.ExpectedReplica, nil, opt.Method, opt.FullTrusted, contractConfig)
+		_, roots, err = uploader.SplitableUpload(ctx, file, uploadArgs.fragmentSize, opt)
 		if err != nil {
-			return nil, nil, err
+			logrus.WithError(err).Fatal("Failed to upload file")
 		}
-
-		return up, indexerClient.Close, nil
 	}
 
-	clients := node.MustNewZgsClients(args.node, nil, providerOption)
-	closer := func() {
-		for _, client := range clients {
-			client.Close()
+	if len(roots) == 1 {
+		logrus.Infof("file uploaded, root = %v", roots[0])
+	} else {
+		s := make([]string, len(roots))
+		for i, root := range roots {
+			s[i] = root.String()
 		}
+		logrus.Infof("file uploaded in %v fragments, roots = %v", len(roots), strings.Join(s, ","))
 	}
-
-	up, err := transfer.NewUploaderWithContractConfig(ctx, w3client, &transfer.SelectedNodes{Trusted: clients}, contractConfig, zg_common.LogOption{Logger: logrus.StandardLogger()})
-	if err != nil {
-		closer()
-		return nil, nil, err
-	}
-
-	return up, closer, nil
 }

cmd/upload_dir.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 
+	zg_common "github.com/0gfoundation/0g-storage-client/common"
 	"github.com/0gfoundation/0g-storage-client/common/blockchain"
 	"github.com/0gfoundation/0g-storage-client/transfer"
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -55,12 +56,20 @@ func uploadDir(*cobra.Command, []string) {
 		FullTrusted:      uploadDirArgs.fullTrusted,
 	}
 
-	uploader, closer, err := newUploader(ctx, 0, uploadDirArgs, w3client, opt)
+	uploader, closer, err := transfer.NewUploaderFromConfig(ctx, w3client, transfer.UploaderConfig{
+		Nodes:          uploadDirArgs.node,
+		ProviderOption: providerOption,
+		LogOption:      zg_common.LogOption{Logger: logrus.StandardLogger()},
+		Contact: &transfer.ContractAddress{
+			FlowAddress:   uploadDirArgs.flowAddress,
+			MarketAddress: uploadDirArgs.marketAddress,
+		},
+		Routines: uploadDirArgs.routines,
+	})
 	if err != nil {
 		logrus.WithError(err).Fatal("Failed to initialize uploader")
 	}
 	defer closer()
-	uploader.WithRoutines(uploadArgs.routines)
 
 	txnHash, rootHash, err := uploader.UploadDir(ctx, uploadDirArgs.file, opt)
 	if err != nil {

gateway/local_apis.go

@@ -123,7 +123,11 @@ func uploadLocalFile(c *gin.Context) (interface{}, error) {
 		return nil, api.ErrValidation.WithData("node index out of bound")
 	}
 
-	uploader, err := transfer.NewUploader(context.Background(), nil, &transfer.SelectedNodes{Trusted: []*node.ZgsClient{allClients[input.Node]}}, zg_common.LogOption{Logger: logrus.StandardLogger()})
+	uploader, err := transfer.NewUploaderWithContractConfig(context.Background(), nil, &transfer.SelectedNodes{
+		Trusted: []*node.ZgsClient{allClients[input.Node]},
+	}, transfer.UploaderConfig{
+		LogOption: zg_common.LogOption{Logger: logrus.StandardLogger()},
+	})
 	if err != nil {
 		return nil, errors.WithMessage(err, "Failed to create uploader")
 	}

indexer/client.go

@@ -38,24 +38,22 @@ type IndexerClientOption struct {
 	ProviderOption providers.Option
 	LogOption      common.LogOption // log option when uploading data
 	FullTrusted    bool             // whether to use full trusted nodes
+	Routines       int              // number of routines for uploader
+	Contract       *transfer.ContractAddress
 }
 
 // NewClient create new indexer client, url is indexer service url
-func NewClient(url string, option ...IndexerClientOption) (*Client, error) {
-	var opt IndexerClientOption
-	if len(option) > 0 {
-		opt = option[0]
-	}
+func NewClient(url string, option IndexerClientOption) (*Client, error) {
 
-	client, err := rpc.NewClient(url, opt.ProviderOption)
+	client, err := rpc.NewClient(url, option.ProviderOption)
 	if err != nil {
 		return nil, err
 	}
 
 	return &Client{
 		Client: client,
-		option: opt,
-		logger: common.NewLogger(opt.LogOption),
+		option: option,
+		logger: common.NewLogger(option.LogOption),
 	}, nil
 }
 
@@ -116,41 +114,69 @@ func (c *Client) SelectNodes(ctx context.Context, expectedReplica uint, dropped
 	}, nil
 }
 
-// NewUploaderFromIndexerNodes return an uploader with selected storage nodes from indexer service.
-func (c *Client) NewUploaderFromIndexerNodes(ctx context.Context, segNum uint64, w3Client *web3go.Client, expectedReplica uint, dropped []string, method string, fullTrusted bool) (*transfer.Uploader, error) {
-	return c.NewUploaderFromIndexerNodesWithContractConfig(ctx, segNum, w3Client, expectedReplica, dropped, method, fullTrusted, nil)
-}
-
 // NewUploaderFromIndexerNodesWithContractConfig returns an uploader with selected storage nodes and optional contract config.
-func (c *Client) NewUploaderFromIndexerNodesWithContractConfig(ctx context.Context, segNum uint64, w3Client *web3go.Client, expectedReplica uint, dropped []string, method string, fullTrusted bool, contractConfig *transfer.UploaderContractConfig) (*transfer.Uploader, error) {
+func (c *Client) NewUploaderFromIndexerNodes(ctx context.Context, segNum uint64, w3Client *web3go.Client, expectedReplica uint, dropped []string, method string, fullTrusted bool) (*transfer.Uploader, error) {
 	selected, err := c.SelectNodes(ctx, expectedReplica, dropped, method, fullTrusted)
 	if err != nil {
 		return nil, err
 	}
 
 	c.logger.Infof("get storage nodes from indexer (trusted: %v, discovered: %v)", len(selected.Trusted), len(selected.Discovered))
-	return transfer.NewUploaderWithContractConfig(ctx, w3Client, selected, contractConfig, c.option.LogOption)
+	uploader, err := transfer.NewUploaderWithContractConfig(ctx, w3Client, selected, transfer.UploaderConfig{
+		Routines:  c.option.Routines,
+		LogOption: c.option.LogOption,
+		Contact:   c.option.Contract,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return uploader, nil
 }
 
-// Upload submit data to 0g storage contract, then transfer the data to the storage nodes selected from indexer service.
-func (c *Client) Upload(ctx context.Context, w3Client *web3go.Client, data core.IterableData, option ...transfer.UploadOption) (eth_common.Hash, error) {
-	expectedReplica := uint(1)
+// SplitableUpload submits data and retries on node errors. If FullTrusted is false,
+// it tries once and falls back to full trusted nodes.
+func (c *Client) SplitableUpload(ctx context.Context, w3Client *web3go.Client, data core.IterableData, fragmentSize int64, option ...transfer.UploadOption) ([]eth_common.Hash, []eth_common.Hash, error) {
+	var opt transfer.UploadOption
 	if len(option) > 0 {
-		expectedReplica = max(expectedReplica, option[0].ExpectedReplica)
+		opt = option[0]
 	}
+	expectedReplica := max(uint(1), opt.ExpectedReplica)
+	maxRetry := opt.NRetries
+	if maxRetry <= 0 {
+		maxRetry = 3
+	}
+
 	dropped := make([]string, 0)
+	attempts := 0
+
 	for {
-		uploader, err := c.NewUploaderFromIndexerNodes(ctx, data.NumSegments(), w3Client, expectedReplica, dropped, option[0].Method, option[0].FullTrusted)
+		uploader, err := c.NewUploaderFromIndexerNodes(ctx, data.NumSegments(), w3Client, expectedReplica, dropped, opt.Method, opt.FullTrusted)
 		if err != nil {
-			return eth_common.Hash{}, err
+			return nil, nil, err
+		}
+
+		txHashes, roots, err := uploader.SplitableUpload(ctx, data, fragmentSize, opt)
+		if err == nil {
+			return txHashes, roots, nil
 		}
-		txHash, _, err := uploader.Upload(ctx, data, option...)
+
+		if !opt.FullTrusted {
+			opt.FullTrusted = true
+			c.logger.WithError(err).Warn("Upload failed, retrying with full trusted nodes")
+		} else {
+			attempts += 1
+		}
+
 		var rpcError *node.RPCError
 		if errors.As(err, &rpcError) {
 			dropped = append(dropped, rpcError.URL)
 			c.logger.Infof("dropped problematic node and retry: %v", rpcError.Error())
 		} else {
-			return txHash, err
+			c.logger.WithError(err).Warn("Upload failed, retrying")
+		}
+
+		if attempts >= maxRetry {
+			return txHashes, roots, err
 		}
 	}
 }

kv/batcher.go

@@ -51,7 +51,9 @@ func (b *Batcher) Exec(ctx context.Context, option ...transfer.UploadOption) (co
 	}
 
 	// upload file
-	uploader, err := transfer.NewUploader(ctx, b.w3Client, b.clients, zg_common.LogOption{Logger: b.logger})
+	uploader, err := transfer.NewUploaderWithContractConfig(ctx, b.w3Client, b.clients, transfer.UploaderConfig{
+		LogOption: zg_common.LogOption{Logger: b.logger},
+	})
 	if err != nil {
 		return common.Hash{}, err
 	}

requirements.txt

@@ -1,10 +1,10 @@
 jsonrpcclient==4.0.3
 pyyaml==6.0.1
-pysha3==1.0.2
-coincurve==18.0.0
-eth-utils==3.0.0
-py-ecc==7.0.0
-web3==6.14.0
+pycryptodome==3.20.0
+coincurve==21.0.0
+eth-utils==5.0.0
+py-ecc==8.0.0
+web3==7.14.0
 eth_tester
-cffi==1.16.0
-rtoml==0.10.0
+cffi==2.0.0
+rtoml==0.13.0