diff --git a/cmd/server/main.go b/cmd/server/main.go index 8077751..a5344b7 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -90,6 +90,8 @@ type Options struct { lagMin int `json:"lag_min,omitempty"` lagMax int `json:"lag_max,omitempty"` lagWindowMin int `json:"lag_window_min,omitempty"` + + rpcTimeout time.Duration `json:"rpc_timeout,omitempty"` } func main() { @@ -108,6 +110,7 @@ func main() { flag.IntVar(&opts.lagMin, "lag-min", 1, "minimum confirmation lag in blocks (applied when the chain is stable)") flag.IntVar(&opts.lagMax, "lag-max", 12, "maximum confirmation lag in blocks (cap during heavy reorgs)") flag.IntVar(&opts.lagWindowMin, "lag-window", 30, "minutes of recent reorg history used to size the adaptive lag") + flag.DurationVar(&opts.rpcTimeout, "rpc-timeout", frontend.DefaultRPCTimeout, "bound a single dragonxd JSON-RPC call; 0 disables. One stuck call otherwise blocks every other caller, because HTTP POST mode serialises all RPC through one goroutine") // creating --version as a requirement of help2man if len(os.Args) > 1 && (os.Args[1] == "--version" || os.Args[1] == "-v") { @@ -172,13 +175,13 @@ func main() { // sending transactions, but in the future it could back a different type // of block streamer. - rpcClient, err := frontend.NewZRPCFromConf(opts.confPath) + rpcClient, err := frontend.NewZRPCFromConf(opts.confPath, opts.rpcTimeout) if err != nil { log.WithFields(logrus.Fields{ "error": err, }).Warn("DRAGONX.conf failed, will try empty credentials for rpc") - rpcClient, err = frontend.NewZRPCFromCreds("127.0.0.1:21769", "", "") + rpcClient, err = frontend.NewZRPCFromCreds("127.0.0.1:21769", "", "", opts.rpcTimeout) if err != nil { log.WithFields(logrus.Fields{ diff --git a/common/common.go b/common/common.go index 24172b4..c53ea11 100644 --- a/common/common.go +++ b/common/common.go @@ -10,12 +10,12 @@ import ( "git.hush.is/hush/lightwalletd/parser" "git.hush.is/hush/lightwalletd/walletrpc" - "github.com/btcsuite/btcd/rpcclient" + "git.hush.is/hush/lightwalletd/zrpc" "github.com/pkg/errors" "github.com/sirupsen/logrus" ) -func GetSaplingInfo(rpcClient *rpcclient.Client) (int, int, string, string, int, int, int, error) { +func GetSaplingInfo(rpcClient *zrpc.Client) (int, int, string, string, int, int, int, error) { result, rpcErr := rpcClient.RawRequest("getblockchaininfo", make([]json.RawMessage, 0)) var err error @@ -79,7 +79,7 @@ func GetSaplingInfo(rpcClient *rpcclient.Client) (int, int, string, string, int, return int(saplingHeight), int(blockHeight), chainName, branchID, int(difficulty), int(longestchain), int(notarized), nil } -func GetCoinsupply(rpcClient *rpcclient.Client) (string, string, int, int, int, int, error) { +func GetCoinsupply(rpcClient *zrpc.Client) (string, string, int, int, int, int, error) { result1, rpcErr := rpcClient.RawRequest("coinsupply", make([]json.RawMessage, 0)) var err error @@ -144,7 +144,7 @@ func GetCoinsupply(rpcClient *rpcclient.Client) (string, string, int, int, int, return result, coin, height, supply, zfunds, total, nil } -func getBlockFromRPC(rpcClient *rpcclient.Client, height int) (*walletrpc.CompactBlock, error) { +func getBlockFromRPC(rpcClient *zrpc.Client, height int) (*walletrpc.CompactBlock, error) { params := make([]json.RawMessage, 2) params[0] = json.RawMessage("\"" + strconv.Itoa(height) + "\"") params[1] = json.RawMessage("0") @@ -187,7 +187,7 @@ func getBlockFromRPC(rpcClient *rpcclient.Client, height int) (*walletrpc.Compac return block.ToCompact(), nil } -func BlockIngestor(rpcClient *rpcclient.Client, cache *BlockCache, log *logrus.Entry, +func BlockIngestor(rpcClient *zrpc.Client, cache *BlockCache, log *logrus.Entry, stopChan chan bool, startHeight int) { reorgCount := 0 height := startHeight @@ -263,7 +263,7 @@ func BlockIngestor(rpcClient *rpcclient.Client, cache *BlockCache, log *logrus.E } } -func GetBlock(rpcClient *rpcclient.Client, cache *BlockCache, height int) (*walletrpc.CompactBlock, error) { +func GetBlock(rpcClient *zrpc.Client, cache *BlockCache, height int) (*walletrpc.CompactBlock, error) { // Don't serve blocks above the advertised (lag-adjusted) tip, so wallets can // neither sync nor anchor shielded spends into the unstable reorg zone. if !cache.HeightAllowed(height) { @@ -293,7 +293,7 @@ func GetBlock(rpcClient *rpcclient.Client, cache *BlockCache, height int) (*wall return block, nil } -func GetBlockRange(rpcClient *rpcclient.Client, cache *BlockCache, +func GetBlockRange(rpcClient *zrpc.Client, cache *BlockCache, blockOut chan<- walletrpc.CompactBlock, errOut chan<- error, start, end int) { // Go over [start, end] inclusive diff --git a/frontend/mempool.go b/frontend/mempool.go index 0d8d302..6245d6e 100644 --- a/frontend/mempool.go +++ b/frontend/mempool.go @@ -6,7 +6,7 @@ import ( "sync" "time" - "github.com/btcsuite/btcd/rpcclient" + "git.hush.is/hush/lightwalletd/zrpc" "github.com/sirupsen/logrus" "git.hush.is/hush/lightwalletd/common" @@ -37,7 +37,7 @@ var ( // getMempoolMonitor returns the process-wide mempool monitor, starting its poller on first use. // The client/cache/log are bound once (they are process singletons on the SqlStreamer). -func getMempoolMonitor(client *rpcclient.Client, cache *common.BlockCache, log *logrus.Entry) *mempoolMonitor { +func getMempoolMonitor(client *zrpc.Client, cache *common.BlockCache, log *logrus.Entry) *mempoolMonitor { mempoolMonitorOnce.Do(func() { sharedMempoolMonitor = &mempoolMonitor{ subs: make(map[int]chan *walletrpc.RawTransaction), @@ -87,7 +87,7 @@ func (m *mempoolMonitor) reset() { } } -func (m *mempoolMonitor) run(client *rpcclient.Client, cache *common.BlockCache, log *logrus.Entry) { +func (m *mempoolMonitor) run(client *zrpc.Client, cache *common.BlockCache, log *logrus.Entry) { lastHeight := cache.GetLatestBlock() for { // Track block height even while idle, so a new subscriber isn't immediately closed by a diff --git a/frontend/rpc_client.go b/frontend/rpc_client.go index 5723e7d..a742639 100644 --- a/frontend/rpc_client.go +++ b/frontend/rpc_client.go @@ -2,13 +2,24 @@ package frontend import ( "net" + "time" - "github.com/btcsuite/btcd/rpcclient" + "git.hush.is/hush/lightwalletd/zrpc" "github.com/pkg/errors" ini "gopkg.in/ini.v1" ) -func NewZRPCFromConf(confPath string) (*rpcclient.Client, error) { +// DefaultRPCTimeout bounds a single JSON-RPC round trip to dragonxd. +// +// Why 120s and not something tighter: the slowest legitimate call this daemon +// makes is `coinsupply`, measured at 48s against a cold UTXO set (3s once the +// node has cached the result). A timeout below that would turn a slow-but- +// working call into a hard failure. 120s leaves ~2.5x headroom over the +// slowest real call while still bounding a hang that is otherwise unbounded -- +// calls were seen running past five minutes before the process was killed. +const DefaultRPCTimeout = 120 * time.Second + +func NewZRPCFromConf(confPath string, timeout time.Duration) (*zrpc.Client, error) { cfg, err := ini.Load(confPath) if err != nil { return nil, errors.Wrap(err, "failed to read config file") @@ -19,19 +30,10 @@ func NewZRPCFromConf(confPath string) (*rpcclient.Client, error) { username := cfg.Section("").Key("rpcuser").String() password := cfg.Section("").Key("rpcpassword").String() - return NewZRPCFromCreds(net.JoinHostPort(rpcaddr, rpcport), username, password) + return NewZRPCFromCreds(net.JoinHostPort(rpcaddr, rpcport), username, password, timeout) } -func NewZRPCFromCreds(addr, username, password string) (*rpcclient.Client, error) { - // Connect to local DragonX RPC server using HTTP POST mode. - connCfg := &rpcclient.ConnConfig{ - Host: addr, - User: username, - Pass: password, - HTTPPostMode: true, // DragonX only supports HTTP POST mode - DisableTLS: true, // DragonX does not provide TLS by default - } - // Notice the notification parameter is nil since notifications are - // not supported in HTTP POST mode. - return rpcclient.New(connCfg, nil) +func NewZRPCFromCreds(addr, username, password string, timeout time.Duration) (*zrpc.Client, error) { + // DragonX only supports HTTP POST mode and does not provide TLS by default. + return zrpc.New(addr, username, password, timeout), nil } diff --git a/frontend/service.go b/frontend/service.go index 608cf60..cdccf2d 100644 --- a/frontend/service.go +++ b/frontend/service.go @@ -10,7 +10,7 @@ import ( "strings" "time" - "github.com/btcsuite/btcd/rpcclient" + "git.hush.is/hush/lightwalletd/zrpc" "github.com/sirupsen/logrus" "git.hush.is/hush/lightwalletd/common" @@ -24,11 +24,11 @@ var ( // the service type type SqlStreamer struct { cache *common.BlockCache - client *rpcclient.Client + client *zrpc.Client log *logrus.Entry } -func NewSQLiteStreamer(client *rpcclient.Client, cache *common.BlockCache, log *logrus.Entry) (walletrpc.CompactTxStreamerServer, error) { +func NewSQLiteStreamer(client *zrpc.Client, cache *common.BlockCache, log *logrus.Entry) (walletrpc.CompactTxStreamerServer, error) { return &SqlStreamer{cache, client, log}, nil } diff --git a/zrpc/client.go b/zrpc/client.go new file mode 100644 index 0000000..ddfd3ee --- /dev/null +++ b/zrpc/client.go @@ -0,0 +1,149 @@ +// Package zrpc is a minimal JSON-RPC client for talking to dragonxd. +// +// It replaces github.com/btcsuite/btcd/rpcclient, of which lightwalletd used +// exactly one method: RawRequest. That package is unusable here for two +// reasons, both of which this package exists to fix: +// +// 1. NO TIMEOUT, AND NO WAY TO SET ONE. rpcclient builds its http.Client in an +// unexported newHTTPClient() and its ConnConfig exposes no Timeout field, so +// a request can hang forever. Calls were observed hanging for over five +// minutes against a healthy node that answered the same query from the CLI +// in 2ms. +// +// 2. EVERY CALL IN THE PROCESS IS SERIALISED. In HTTP POST mode rpcclient runs +// a single sendPostHandler goroutine which invokes handleSendPostMessage +// SYNCHRONOUSLY, so at most one RPC is ever in flight. Combined with (1), +// one stuck call blocks the block ingestor, the mempool monitor and every +// user-facing gRPC handler indefinitely. A timeout alone would not fix this: +// bounding the caller's wait still leaves the shared goroutine stuck on +// http.Client.Do, so everything queued behind it stays blocked. Only a +// timeout on the HTTP client itself aborts the in-flight request, and only +// dropping the shared goroutine lets independent callers proceed. +// +// The wire format, request envelope, ID sequence and error semantics are +// deliberately byte-identical to rpcclient's RawRequest. In particular +// RPCError.Error() must render as ": ", because callers parse +// the code back out of the string (see common.GetSaplingInfo, which checks for +// code -8 via strings.SplitN(err.Error(), ":", 2)). +package zrpc + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io/ioutil" + "net/http" + "sync/atomic" + "time" +) + +// RPCError is a JSON-RPC error object returned by dragonxd. +type RPCError struct { + Code int64 `json:"code,omitempty"` + Message string `json:"message,omitempty"` +} + +// Error renders as ": ". Callers depend on this exact shape to +// recover the numeric code; do not change it. +func (e *RPCError) Error() string { + return fmt.Sprintf("%d: %s", e.Code, e.Message) +} + +type request struct { + Jsonrpc string `json:"jsonrpc"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` + ID int64 `json:"id"` +} + +type rawResponse struct { + Result json.RawMessage `json:"result"` + Error *RPCError `json:"error"` +} + +// Client is safe for concurrent use by multiple goroutines. +type Client struct { + url string + user string + pass string + http *http.Client + nextID int64 +} + +// New returns a client for a dragonxd JSON-RPC endpoint. A timeout of 0 means +// no timeout, which reproduces the old unbounded behaviour and should not be +// used in production. +func New(addr, user, pass string, timeout time.Duration) *Client { + return &Client{ + url: "http://" + addr, + user: user, + pass: pass, + http: &http.Client{ + // Covers the whole exchange: connect, write, response headers and + // body read. This is the bound that was missing. + Timeout: timeout, + Transport: &http.Transport{ + // dragonxd's HTTP server supports keep-alive. rpcclient set + // Close=true and opened a fresh TCP connection per request, + // which left hundreds of sockets in TIME_WAIT on a busy node. + MaxIdleConns: 32, + MaxIdleConnsPerHost: 32, + IdleConnTimeout: 90 * time.Second, + }, + }, + } +} + +// RawRequest sends a JSON-RPC request and returns the raw result. A JSON-RPC +// error from the server is returned as *RPCError. +func (c *Client) RawRequest(method string, params []json.RawMessage) (json.RawMessage, error) { + if method == "" { + return nil, errors.New("no method") + } + // Marshal parameters as "[]" instead of "null" when none are passed. + if params == nil { + params = []json.RawMessage{} + } + + body, err := json.Marshal(&request{ + Jsonrpc: "1.0", + Method: method, + Params: params, + ID: atomic.AddInt64(&c.nextID, 1), + }) + if err != nil { + return nil, err + } + + httpReq, err := http.NewRequest("POST", c.url, bytes.NewReader(body)) + if err != nil { + return nil, err + } + httpReq.Header.Set("Content-Type", "application/json") + httpReq.SetBasicAuth(c.user, c.pass) + + httpResp, err := c.http.Do(httpReq) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + respBytes, err := ioutil.ReadAll(httpResp.Body) + if err != nil { + return nil, fmt.Errorf("error reading json reply: %v", err) + } + + var resp rawResponse + if err := json.Unmarshal(respBytes, &resp); err != nil { + // Not a valid JSON-RPC response: surface the status and raw body, the + // same way rpcclient did. dragonxd returns non-JSON bodies for some + // auth and workqueue failures, and callers log this verbatim. + return nil, fmt.Errorf("status code: %d, response: %q", + httpResp.StatusCode, string(respBytes)) + } + if resp.Error != nil { + return nil, resp.Error + } + return resp.Result, nil +} diff --git a/zrpc/live_test.go b/zrpc/live_test.go new file mode 100644 index 0000000..270ac12 --- /dev/null +++ b/zrpc/live_test.go @@ -0,0 +1,86 @@ +package zrpc + +import ( + "encoding/json" + "os" + "strconv" + "strings" + "testing" + "time" + + ini "gopkg.in/ini.v1" +) + +// Live tests against a local dragonxd. Skipped unless ZRPC_CONF points at a +// DRAGONX.conf, so `go test ./...` stays hermetic. +func liveClient(t *testing.T, timeout time.Duration) *Client { + t.Helper() + conf := os.Getenv("ZRPC_CONF") + if conf == "" { + t.Skip("ZRPC_CONF not set; skipping live RPC test") + } + cfg, err := ini.Load(conf) + if err != nil { + t.Fatalf("load conf: %v", err) + } + k := func(n string) string { return cfg.Section("").Key(n).String() } + return New(k("rpcbind")+":"+k("rpcport"), k("rpcuser"), k("rpcpassword"), timeout) +} + +func TestLiveSuccess(t *testing.T) { + c := liveClient(t, 30*time.Second) + res, err := c.RawRequest("getblockchaininfo", nil) + if err != nil { + t.Fatalf("getblockchaininfo: %v", err) + } + var f map[string]interface{} + if err := json.Unmarshal(res, &f); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if f["chain"] != "main" { + t.Fatalf("chain = %v, want main", f["chain"]) + } + t.Logf("ok: chain=%v blocks=%v", f["chain"], f["blocks"]) +} + +// The error string must stay ": " -- common.GetSaplingInfo +// recovers the numeric code with strings.SplitN(err.Error(), ":", 2). +func TestLiveErrorStringShape(t *testing.T) { + c := liveClient(t, 30*time.Second) + p := []json.RawMessage{json.RawMessage(`"99999999"`)} + _, err := c.RawRequest("getblock", p) + if err == nil { + t.Fatal("expected an error for an out-of-range height") + } + parts := strings.SplitN(err.Error(), ":", 2) + code, perr := strconv.ParseInt(parts[0], 10, 32) + if perr != nil { + t.Fatalf("error string %q does not start with a numeric code", err.Error()) + } + if code != -8 { + t.Logf("note: code %d (expected -8 for a bad height, but any numeric code proves the shape)", code) + } + t.Logf("ok: %q -> code %d", err.Error(), code) +} + +// A timeout must actually abort the call rather than hanging. +func TestLiveTimeoutFires(t *testing.T) { + c := liveClient(t, 1*time.Nanosecond) + start := time.Now() + _, err := c.RawRequest("getblockchaininfo", nil) + elapsed := time.Since(start) + if err == nil { + t.Fatal("expected a timeout error") + } + if elapsed > 5*time.Second { + t.Fatalf("timeout did not fire promptly: %v", elapsed) + } + t.Logf("ok: timed out in %v with %v", elapsed, err) +} + +func TestNoMethod(t *testing.T) { + c := New("127.0.0.1:1", "u", "p", time.Second) + if _, err := c.RawRequest("", nil); err == nil || err.Error() != "no method" { + t.Fatalf(`RawRequest("") = %v, want "no method"`, err) + } +}