Skip to content

Add failover support for multiple RPC nodes #162

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
TheMarstonConnell wants to merge 5 commits into
base: main
Choose a base branch
from
+593 −113
Open

Add failover support for multiple RPC nodes #162

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8d950eb
Add RPC/GRPC node failover support
claude Jan 2, 2026
574d4eb
Fix potential lint issues in failover.go
claude Jan 2, 2026
33ca05c
Refactor failover.go to address code review suggestions
claude Jan 2, 2026
7d28ec4
Address CodeRabbit review suggestions
claude Jan 2, 2026
eb285bb
Address additional CodeRabbit review suggestions
claude Jan 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions api/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,16 @@ import (
"net/http"

"github.com/JackalLabs/sequoia/api/types"
"github.com/JackalLabs/sequoia/rpc"
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/desmos-labs/cosmos-go-wallet/client"
storageTypes "github.com/jackalLabs/canine-chain/v5/x/storage/types"
"github.com/rs/zerolog/log"
)

func SpaceHandler(c *client.Client, address string) func(http.ResponseWriter, *http.Request) {
func SpaceHandler(fc *rpc.FailoverClient) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
queryClient := storageTypes.NewQueryClient(c.GRPCConn)
queryClient := storageTypes.NewQueryClient(fc.GRPCConn())
address := fc.AccAddress()

params := &storageTypes.QueryProvider{
Address: address,
Expand Down
31 changes: 15 additions & 16 deletions api/file_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import (
"time"

"github.com/JackalLabs/sequoia/api/gateway"
"github.com/JackalLabs/sequoia/rpc"
sequoiaTypes "github.com/JackalLabs/sequoia/types"

"github.com/JackalLabs/sequoia/utils"
Expand All @@ -23,8 +24,6 @@ import (

"github.com/JackalLabs/sequoia/proofs"

"github.com/desmos-labs/cosmos-go-wallet/wallet"

"github.com/JackalLabs/sequoia/api/types"
"github.com/JackalLabs/sequoia/file_system"
"github.com/gorilla/mux"
Expand All @@ -45,7 +44,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, fc *rpc.FailoverClient, chunkSize int64) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
// Use streaming multipart parsing instead of loading entire form into memory
sender, merkleString, startBlockString, proofTypeString, file, _, err := parseMultipartFormStreaming(req)
Expand Down Expand Up @@ -80,7 +79,7 @@ func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wal
// Size validation is now enforced during multipart streaming
// Files larger than MaxFileSize (32GB) will be rejected immediately

cl := storageTypes.NewQueryClient(wl.Client.GRPCConn)
cl := storageTypes.NewQueryClient(fc.GRPCConn())
queryParams := storageTypes.QueryFile{
Merkle: merkle,
Owner: sender,
Expand All @@ -100,7 +99,7 @@ func PostFileHandler(fio *file_system.FileSystem, prover *proofs.Prover, wl *wal
}

if len(f.Proofs) == int(f.MaxProofs) {
if !f.ContainsProver(wl.AccAddress()) {
if !f.ContainsProver(fc.AccAddress()) {
handleErr(fmt.Errorf("cannot accept file that I cannot claim"), w, http.StatusInternalServerError)
return
}
Expand Down Expand Up @@ -133,7 +132,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, fc *rpc.FailoverClient, chunkSize int64) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
// Use streaming multipart parsing instead of loading entire form into memory
sender, merkleString, startBlockString, _, file, _, err := parseMultipartFormStreaming(req)
Expand Down Expand Up @@ -187,7 +186,7 @@ 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)
cl := storageTypes.NewQueryClient(fc.GRPCConn())
queryParams := storageTypes.QueryFile{
Merkle: merkle,
Owner: sender,
Expand All @@ -211,7 +210,7 @@ func PostFileHandlerV2(fio *file_system.FileSystem, prover *proofs.Prover, wl *w
}

if len(f.Proofs) == int(f.MaxProofs) {
if !f.ContainsProver(wl.AccAddress()) {
if !f.ContainsProver(fc.AccAddress()) {
log.Error().Err(fmt.Errorf("cannot accept file that I cannot claim"))
up.Status = "Error: Can't claim"
return
Expand Down Expand Up @@ -413,7 +412,7 @@ func getFolderData(data io.Reader) (*sequoiaTypes.FolderData, bool) {

// getMerkleData retrieves file data by merkle hash, first attempting local storage and then querying network providers if not found locally.
// It returns the file data if successful, or an error if the file cannot be retrieved from any source.
func getMerkleData(merkle []byte, fileName string, f *file_system.FileSystem, wallet *wallet.Wallet, myIp string) (io.ReadSeekCloser, error) {
func getMerkleData(merkle []byte, fileName string, f *file_system.FileSystem, fc *rpc.FailoverClient, myIp string) (io.ReadSeekCloser, error) {
file, err := f.GetFileData(merkle)
if err == nil {
return file, nil
Expand All @@ -425,7 +424,7 @@ func getMerkleData(merkle []byte, fileName string, f *file_system.FileSystem, wa
Merkle: merkle,
}

cl := storageTypes.NewQueryClient(wallet.Client.GRPCConn)
cl := storageTypes.NewQueryClient(fc.GRPCConn())

res, err := cl.FindFile(context.Background(), queryParams)
if err != nil {
Expand Down Expand Up @@ -470,10 +469,10 @@ func getMerkleData(merkle []byte, fileName string, f *file_system.FileSystem, wa
// GetMerklePathData recursively resolves a file or folder by traversing a path from a root merkle hash.
// If the path leads to a file, returns its data; if it leads to a folder and raw is false, returns an HTML representation of the folder.
// Returns the file or folder data, the resolved filename, and an error if the path is invalid or data retrieval fails.
func GetMerklePathData(root []byte, path []string, fileName string, f *file_system.FileSystem, wallet *wallet.Wallet, myIp string, currentPath string, raw bool) (io.ReadSeekCloser, string, error) {
func GetMerklePathData(root []byte, path []string, fileName string, f *file_system.FileSystem, fc *rpc.FailoverClient, myIp string, currentPath string, raw bool) (io.ReadSeekCloser, string, error) {
currentRoot := root

fileData, err := getMerkleData(currentRoot, fileName, f, wallet, myIp)
fileData, err := getMerkleData(currentRoot, fileName, f, fc, myIp)
if err != nil {
return nil, fileName, err
}
Expand All @@ -491,7 +490,7 @@ func GetMerklePathData(root []byte, path []string, fileName string, f *file_syst

for _, child := range children {
if child.Name == p {
return GetMerklePathData(child.Merkle, path[1:], child.Name, f, wallet, myIp, currentPath, raw) // check the next item in the list
return GetMerklePathData(child.Merkle, path[1:], child.Name, f, fc, myIp, currentPath, raw) // check the next item in the list
}
}
// did not find child
Expand Down Expand Up @@ -522,7 +521,7 @@ func GetMerklePathData(root []byte, path []string, fileName string, f *file_syst
// FindFileHandler returns an HTTP handler that serves files or folders by merkle hash and optional path, supporting raw or HTML folder views.
//
// The handler extracts the merkle hash and optional path from the request, resolves the requested file or folder (recursively if a path is provided), and serves the content. If the target is a folder and the `raw` query parameter is not set, an HTML representation is generated. If a filename is not specified, the merkle string is used as the default name. Errors are returned as JSON responses.
func FindFileHandler(f *file_system.FileSystem, wallet *wallet.Wallet, myIp string) func(http.ResponseWriter, *http.Request) {
func FindFileHandler(f *file_system.FileSystem, fc *rpc.FailoverClient, myIp string) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
vars := mux.Vars(req)
fileName := req.URL.Query().Get("filename")
Expand Down Expand Up @@ -555,7 +554,7 @@ func FindFileHandler(f *file_system.FileSystem, wallet *wallet.Wallet, myIp stri
}

if len(filteredPaths) > 0 {
data, name, err := GetMerklePathData(merkle, filteredPaths, fileName, f, wallet, myIp, req.URL.Path, raw)
data, name, err := GetMerklePathData(merkle, filteredPaths, fileName, f, fc, myIp, req.URL.Path, raw)
if err != nil {
v := types.ErrorResponse{
Error: err.Error(),
Expand All @@ -572,7 +571,7 @@ func FindFileHandler(f *file_system.FileSystem, wallet *wallet.Wallet, myIp stri

// This code will only run if there's no path or the path is empty

fileData, err := getMerkleData(merkle, fileName, f, wallet, myIp)
fileData, err := getMerkleData(merkle, fileName, f, fc, myIp)
if err != nil {
v := types.ErrorResponse{
Error: err.Error(),
Expand Down
12 changes: 6 additions & 6 deletions api/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,9 @@
"net/http"

"github.com/JackalLabs/sequoia/config"
"github.com/JackalLabs/sequoia/rpc"

"github.com/JackalLabs/sequoia/api/types"
"github.com/desmos-labs/cosmos-go-wallet/wallet"
"github.com/rs/zerolog/log"
)

Expand All @@ -26,9 +26,9 @@
}
}

func VersionHandler(wallet *wallet.Wallet) func(http.ResponseWriter, *http.Request) {
func VersionHandler(fc *rpc.FailoverClient) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
chainId, err := wallet.Client.GetChainID()
chainId, err := fc.Wallet().Client.GetChainID()
if err != nil {
w.WriteHeader(500)
return
Expand All @@ -48,15 +48,15 @@
}
}

func NetworkHandler(wallet *wallet.Wallet) func(http.ResponseWriter, *http.Request) {
func NetworkHandler(fc *rpc.FailoverClient) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
status, err := wallet.Client.RPCClient.Status(context.Background())
status, err := fc.RPCClient().Status(context.Background())
if err != nil {
w.WriteHeader(500)
return
}

grpcStatus := wallet.Client.GRPCConn.GetState()
grpcStatus := fc.GRPCConn().GetState()

Check failure on line 59 in api/index.go

View workflow job for this annotation

GitHub Actions / cov 

fc.GRPCConn().GetState undefined (type "github.com/cosmos/gogoproto/grpc".ClientConn has no field or method GetState)

Check failure on line 59 in api/index.go

View workflow job for this annotation

GitHub Actions / Test 

fc.GRPCConn().GetState undefined (type "github.com/cosmos/gogoproto/grpc".ClientConn has no field or method GetState)

Check failure on line 59 in api/index.go

View workflow job for this annotation

GitHub Actions / Build 

fc.GRPCConn().GetState undefined (type "github.com/cosmos/gogoproto/grpc".ClientConn has no field or method GetState)

Check failure on line 59 in api/index.go

View workflow job for this annotation

GitHub Actions / lint 

fc.GRPCConn().GetState undefined (type "github.com/cosmos/gogoproto/grpc".ClientConn has no field or method GetState) (typecheck)

v := types.NetworkResponse{
GRPCStatus: grpcStatus.String(),
Expand Down
20 changes: 10 additions & 10 deletions api/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,12 @@ import (
"github.com/rs/cors"

"github.com/JackalLabs/sequoia/file_system"
"github.com/JackalLabs/sequoia/rpc"
"github.com/prometheus/client_golang/prometheus/promhttp"

"github.com/JackalLabs/sequoia/proofs"
"github.com/rs/zerolog/log"

"github.com/desmos-labs/cosmos-go-wallet/wallet"
"github.com/gorilla/mux"

jsoniter "github.com/json-iterator/go"
Expand Down Expand Up @@ -47,29 +47,29 @@ 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, fc *rpc.FailoverClient, chunkSize int64, myIp string) {
defer log.Info().Msg("API module stopped")
r := mux.NewRouter()

outline := types.NewOutline()

outline.RegisterGetRoute(r, "/", IndexHandler(wallet.AccAddress()))
outline.RegisterGetRoute(r, "/", IndexHandler(fc.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, fc, chunkSize))
outline.RegisterPostRoute(r, "/v2/upload", PostFileHandlerV2(f, p, fc, chunkSize))
outline.RegisterPostRoute(r, "/v2/status/{id}", CheckUploadStatus())
outline.RegisterPostRoute(r, "/api/jobs", ListJobsHandler())
outline.RegisterGetRoute(r, "/download/{merkle}", DownloadFileHandler(f))

if a.cfg.OpenGateway {
outline.RegisterGetRoute(r, "/get/{merkle}/{path:.*}", FindFileHandler(f, wallet, myIp))
outline.RegisterGetRoute(r, "/get/{merkle}", FindFileHandler(f, wallet, myIp))
outline.RegisterGetRoute(r, "/get/{merkle}/{path:.*}", FindFileHandler(f, fc, myIp))
outline.RegisterGetRoute(r, "/get/{merkle}", FindFileHandler(f, fc, myIp))
}

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(fc))

outline.RegisterGetRoute(r, "/ipfs/peers", IPFSListPeers(f))
outline.RegisterGetRoute(r, "/ipfs/hosts", IPFSListHosts(f))
Expand All @@ -79,8 +79,8 @@ func (a *API) Serve(f *file_system.FileSystem, p *proofs.Prover, wallet *wallet.

// outline.RegisterGetRoute(r, "/dump", DumpDBHandler(f))

outline.RegisterGetRoute(r, "/version", VersionHandler(wallet))
outline.RegisterGetRoute(r, "/network", NetworkHandler(wallet))
outline.RegisterGetRoute(r, "/version", VersionHandler(fc))
outline.RegisterGetRoute(r, "/network", NetworkHandler(fc))

outline.RegisterGetRoute(r, "/api", outline.OutlineHandler())

Expand Down
8 changes: 4 additions & 4 deletions cmd/wallet/wallet.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,16 +65,16 @@ func balanceCMD() *cobra.Command {
return err
}

wallet, err := config.InitWallet(home)
fc, err := config.InitWallet(home)
if err != nil {
return err
}

queryClient := bankTypes.NewQueryClient(wallet.Client.GRPCConn)
queryClient := bankTypes.NewQueryClient(fc.GRPCConn())

params := &bankTypes.QueryBalanceRequest{
Denom: "ujkl",
Address: wallet.AccAddress(),
Address: fc.AccAddress(),
}

res, err := queryClient.Balance(context.Background(), params)
Expand Down Expand Up @@ -125,7 +125,7 @@ func withdrawCMD() *cobra.Command {
&m,
).WithGasAuto().WithFeeAuto()

res, err := wallet.BroadcastTxCommit(data)
res, err := wallet.Wallet().BroadcastTxCommit(data)
if err != nil {
return err
}
Expand Down
47 changes: 39 additions & 8 deletions config/types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,44 @@ type Seed struct {
DerivationPath string `json:"derivation_path"`
}

// required for the mapstructure tag
// ChainConfig holds the configuration for connecting to a blockchain node.
// It supports both single-node (RPCAddr/GRPCAddr) and multi-node failover
// (RPCAddrs/GRPCAddrs) configurations. If the array fields are set, they
// take precedence over the single address fields.
type ChainConfig struct {
Bech32Prefix string `yaml:"bech32_prefix" mapstructure:"bech32_prefix"`
RPCAddr string `yaml:"rpc_addr" mapstructure:"rpc_addr"`
GRPCAddr string `yaml:"grpc_addr" mapstructure:"grpc_addr"`
GasPrice string `yaml:"gas_price" mapstructure:"gas_price"`
GasAdjustment float64 `yaml:"gas_adjustment" mapstructure:"gas_adjustment"`
Bech32Prefix string `yaml:"bech32_prefix" mapstructure:"bech32_prefix"`
RPCAddr string `yaml:"rpc_addr" mapstructure:"rpc_addr"`
GRPCAddr string `yaml:"grpc_addr" mapstructure:"grpc_addr"`
RPCAddrs []string `yaml:"rpc_addrs" mapstructure:"rpc_addrs"`
GRPCAddrs []string `yaml:"grpc_addrs" mapstructure:"grpc_addrs"`
GasPrice string `yaml:"gas_price" mapstructure:"gas_price"`
GasAdjustment float64 `yaml:"gas_adjustment" mapstructure:"gas_adjustment"`
}

// GetRPCAddrs returns the list of RPC addresses to use.
// If RPCAddrs is set, it returns that. Otherwise, it returns a single-element
// slice containing RPCAddr for backward compatibility.
func (c ChainConfig) GetRPCAddrs() []string {
if len(c.RPCAddrs) > 0 {
return c.RPCAddrs
}
if c.RPCAddr != "" {
return []string{c.RPCAddr}
}
return []string{"http://localhost:26657"}
}

// GetGRPCAddrs returns the list of GRPC addresses to use.
// If GRPCAddrs is set, it returns that. Otherwise, it returns a single-element
// slice containing GRPCAddr for backward compatibility.
func (c ChainConfig) GetGRPCAddrs() []string {
if len(c.GRPCAddrs) > 0 {
return c.GRPCAddrs
}
if c.GRPCAddr != "" {
return []string{c.GRPCAddr}
}
return []string{"localhost:9090"}
}

type Config struct {
Expand Down Expand Up @@ -171,8 +202,8 @@ func (c Config) MarshalZerologObject(e *zerolog.Event) {
Int64("StrayCheckInterval", c.StrayManagerCfg.CheckInterval).
Int64("StrayRefreshInterval", c.StrayManagerCfg.RefreshInterval).
Int("StrayHandCount", c.StrayManagerCfg.HandCount).
Str("ChainRPCAddr", c.ChainCfg.RPCAddr).
Str("ChainGRPCAddr", c.ChainCfg.GRPCAddr).
Strs("ChainRPCAddrs", c.ChainCfg.GetRPCAddrs()).
Strs("ChainGRPCAddrs", c.ChainCfg.GetGRPCAddrs()).
Str("ChainGasPrice", c.ChainCfg.GasPrice).
Float64("ChainGasAdjustment", c.ChainCfg.GasAdjustment).
Str("IP", c.Ip).
Expand Down
Loading
Loading