Multithreading queue

.gitignore

@@ -5,3 +5,5 @@ data
 sequoia
 *.txt
 *.json
+~
+.DS_Store

api/client.go

@@ -11,10 +11,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func SpaceHandler(c *client.Client, address string) func(http.ResponseWriter, *http.Request) {
+func SpaceHandler(c *client.Client, queryClient storageTypes.QueryClient, address string) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, req *http.Request) {
-		queryClient := storageTypes.NewQueryClient(c.GRPCConn)
-
 		params := &storageTypes.QueryProvider{
 			Address: address,
 		}

api/file_handler.go

@@ -48,7 +48,7 @@ func handleErr(err error, w http.ResponseWriter, code int) {
 	}
 }
 
-func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wallet.Wallet, chunkSize int64) func(http.ResponseWriter, *http.Request) {
+func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wallet.Wallet, query storageTypes.QueryClient, chunkSize int64) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, req *http.Request) {
 		err := req.ParseMultipartForm(MaxFileSize) // MAX file size lives here
 		if err != nil {
@@ -96,13 +96,12 @@ func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wal
 			return
 		}
 
-		cl := storageTypes.NewQueryClient(wl.Client.GRPCConn)
 		queryParams := storageTypes.QueryFile{
 			Merkle: merkle,
 			Owner:  sender,
 			Start:  startBlock,
 		}
-		res, err := cl.File(context.Background(), &queryParams)
+		res, err := query.File(context.Background(), &queryParams)
 		if err != nil {
 			handleErr(fmt.Errorf("failed to find file on chain with merkle: %x, owner: %s, start: %d | %w", merkle, sender, startBlock, err), w, http.StatusInternalServerError)
 			return
@@ -154,7 +153,7 @@ func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wal
 	}
 }
 
-func PostFileHandlerV2(fio *file_system.FileSystem, prover *proofs.Prover, wl *wallet.Wallet, chunkSize int64) func(http.ResponseWriter, *http.Request) {
+func PostFileHandlerV2(fio *file_system.FileSystem, prover *proofs.Prover, wl *wallet.Wallet, queryClient storageTypes.QueryClient, chunkSize int64) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, req *http.Request) {
 		err := req.ParseMultipartForm(MaxFileSize) // MAX file size lives here
 		if err != nil {
@@ -227,13 +226,12 @@ func PostFileHandlerV2(fio *file_system.FileSystem, prover *proofs.Prover, wl *w
 			log.Error().Err(fmt.Errorf("can't encode json : %w", err))
 		}
 
-		cl := storageTypes.NewQueryClient(wl.Client.GRPCConn)
 		queryParams := storageTypes.QueryFile{
 			Merkle: merkle,
 			Owner:  sender,
 			Start:  startBlock,
 		}
-		res, err := cl.File(context.Background(), &queryParams)
+		res, err := queryClient.File(context.Background(), &queryParams)
 		if err != nil {
 			log.Error().Err(fmt.Errorf("failed to find file on chain with merkle: %x, owner: %s, start: %d | %w", merkle, sender, startBlock, err))
 			return

api/server.go

@@ -13,6 +13,7 @@ import (
 	"github.com/rs/cors"
 
 	"github.com/JackalLabs/sequoia/file_system"
+	storageTypes "github.com/jackalLabs/canine-chain/v4/x/storage/types"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"github.com/JackalLabs/sequoia/proofs"
@@ -47,16 +48,16 @@ func (a *API) Close() error {
 	return a.srv.Close()
 }
 
-func (a *API) Serve(f *file_system.FileSystem, p *proofs.Prover, wallet *wallet.Wallet, chunkSize int64, myIp string) {
+func (a *API) Serve(f *file_system.FileSystem, p *proofs.Prover, wallet *wallet.Wallet, queryClient storageTypes.QueryClient, myIp string, chunkSize int64) {
 	defer log.Info().Msg("API module stopped")
 	r := mux.NewRouter()
 
 	outline := types.NewOutline()
 
 	outline.RegisterGetRoute(r, "/", IndexHandler(wallet.AccAddress()))
 
-	outline.RegisterPostRoute(r, "/upload", PostFileHandler(f, p, wallet, chunkSize))
-	outline.RegisterPostRoute(r, "/v2/upload", PostFileHandlerV2(f, p, wallet, chunkSize))
+	outline.RegisterPostRoute(r, "/upload", PostFileHandler(f, p, wallet, queryClient, chunkSize))
+	outline.RegisterPostRoute(r, "/v2/upload", PostFileHandlerV2(f, p, wallet, queryClient, chunkSize))
 	outline.RegisterPostRoute(r, "/v2/status/{id}", CheckUploadStatus())
 	outline.RegisterPostRoute(r, "/api/jobs", ListJobsHandler())
 	outline.RegisterGetRoute(r, "/download/{merkle}", DownloadFileHandler(f))
@@ -69,7 +70,7 @@ func (a *API) Serve(f *file_system.FileSystem, p *proofs.Prover, wallet *wallet.
 	outline.RegisterGetRoute(r, "/list", ListFilesHandler(f))
 	outline.RegisterGetRoute(r, "/api/client/list", ListFilesHandler(f))
 	outline.RegisterGetRoute(r, "/api/data/fids", LegacyListFilesHandler(f))
-	outline.RegisterGetRoute(r, "/api/client/space", SpaceHandler(wallet.Client, wallet.AccAddress()))
+	outline.RegisterGetRoute(r, "/api/client/space", SpaceHandler(wallet.Client, queryClient, wallet.AccAddress()))
 
 	outline.RegisterGetRoute(r, "/ipfs/peers", IPFSListPeers(f))
 	outline.RegisterGetRoute(r, "/ipfs/hosts", IPFSListHosts(f))
@@ -99,11 +100,18 @@ func (a *API) Serve(f *file_system.FileSystem, p *proofs.Prover, wallet *wallet.
 	}
 
 	log.Logger.Info().Msg(fmt.Sprintf("Sequoia API now listening on %s", a.srv.Addr))
-	err := a.srv.ListenAndServe()
-	if err != nil {
-		if !errors.Is(err, http.ErrServerClosed) {
-			log.Warn().Err(err)
-			return
-		}
+
+	// Create a channel to listen for errors coming from the listener.
+	serverErrors := make(chan error, 1)
+
+	go func() {
+		serverErrors <- a.srv.ListenAndServe()
+	}()
+
+	// Wait for server error
+	err := <-serverErrors
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		log.Warn().Err(err).Msg("server error")
+		return
 	}
 }

cmd/start.go

@@ -45,7 +45,13 @@ func StartCmd() *cobra.Command {
 				log.Logger = log.Level(zerolog.ErrorLevel)
 			}
 
-			app, err := core.NewApp(home)
+			var opts []core.Option
+
+			if cmd.Flags().Changed("test_mode") {
+				opts = append(opts, core.WithTestMode())
+			}
+
+			app, err := core.NewApp(home, opts...)
 			if err != nil {
 				return err
 			}
@@ -61,5 +67,7 @@ func StartCmd() *cobra.Command {
 		},
 	}
 
+	cmd.Flags().Bool("test_mode", false, "run provider without chain network")
+
 	return cmd
 }

config/types.go

@@ -23,8 +23,8 @@ type ChainConfig struct {
 }
 
 type Config struct {
-	QueueInterval    int64              `yaml:"queue_interval" mapstructure:"queue_interval"`
 	ProofInterval    int64              `yaml:"proof_interval" mapstructure:"proof_interval"`
+	QueueConfig      QueueConfig        `yaml:"queue_config" mapstructure:"queue_config"`
 	StrayManagerCfg  StrayManagerConfig `yaml:"stray_manager" mapstructure:"stray_manager"`
 	ChainCfg         ChainConfig        `yaml:"chain_config" mapstructure:"chain_config"`
 	Ip               string             `yaml:"domain" mapstructure:"domain"`
@@ -35,10 +35,6 @@ type Config struct {
 	BlockStoreConfig BlockStoreConfig   `yaml:"block_store_config" mapstructure:"block_store_config"`
 }
 
-func DefaultQueueInterval() int64 {
-	return 10
-}
-
 func DefaultProofInterval() int64 {
 	return 120
 }
@@ -113,7 +109,7 @@ func DefaultBlockStoreConfig() BlockStoreConfig {
 	if err != nil {
 		panic(err)
 	}
-	k, err := priv.Raw()
+	k, err := crypto.MarshalPrivateKey(priv)
 	if err != nil {
 		panic(err)
 	}
@@ -144,10 +140,32 @@ func DefaultChainConfig() ChainConfig {
 	}
 }
 
+type QueueConfig struct {
+	// seconds
+	QueueInterval int64 `yaml:"queue_interval" mapstructure:"queue_interval"`
+	// seconds
+	TxTimer      int16 `yaml:"tx_timer" mapstructure:"tx_timer"`
+	QueueThreads int8  `yaml:"queue_threads" mapstructure:"queue_threads"`
+	// resend tx if network isn't responding
+	MaxRetryAttempt int8 `yaml:"max_retry_attempt" mapstructure:"max_retry_attempt"`
+	// group individual messages into one tx
+	TxBatchSize int8 `yaml:"tx_batch_size" mapstructure:"tx_batch_size"`
+}
+
+func DefaultQueueConfig() QueueConfig {
+	return QueueConfig{
+		QueueInterval:   10,
+		QueueThreads:    5,
+		MaxRetryAttempt: 100,
+		TxBatchSize:     45,
+		TxTimer:         3,
+	}
+}
+
 func DefaultConfig() *Config {
 	return &Config{
-		QueueInterval:    DefaultQueueInterval(),
 		ProofInterval:    DefaultProofInterval(),
+		QueueConfig:      DefaultQueueConfig(),
 		StrayManagerCfg:  DefaultStrayManagerConfig(),
 		ChainCfg:         DefaultChainConfig(),
 		Ip:               DefaultIP(),
@@ -160,8 +178,7 @@ func DefaultConfig() *Config {
 }
 
 func (c Config) MarshalZerologObject(e *zerolog.Event) {
-	e.Int64("QueueInterval", c.QueueInterval).
-		Int64("ProofInterval", c.ProofInterval).
+	e.Int64("ProofInterval", c.ProofInterval).
 		Int64("StrayCheckInterval", c.StrayManagerCfg.CheckInterval).
 		Int64("StrayRefreshInterval", c.StrayManagerCfg.RefreshInterval).
 		Int("StrayHandCount", c.StrayManagerCfg.HandCount).
@@ -180,7 +197,7 @@ func (c Config) MarshalZerologObject(e *zerolog.Event) {
 }
 
 func init() {
-	viper.SetDefault("QueueInterval", DefaultQueueInterval())
+	viper.SetDefault("QueueConfig", DefaultQueueConfig())
 	viper.SetDefault("ProofInterval", DefaultProofInterval())
 	viper.SetDefault("StrayManagerCfg", DefaultStrayManagerConfig())
 	viper.SetDefault("ChainCfg", DefaultChainConfig())