From 4e7a1c0f9bb5822200252e763ffd67d1431b6e7f Mon Sep 17 00:00:00 2001 From: DragonX Developers Date: Wed, 26 Aug 2026 01:29:27 -0500 Subject: [PATCH] rpc: bound every dragonxd call with a timeout, and stop serialising them lightwalletd used github.com/btcsuite/btcd/rpcclient for exactly one method, RawRequest, and that package has two defects that together produce the failure seen on the primary node: 1. It has no timeout and no way to set one. The http.Client is built in an unexported newHTTPClient() and ConnConfig exposes no Timeout field, so a call can hang forever. Calls were observed running past five minutes against a healthy node that answered the same query from dragonx-cli in 2ms. 2. In HTTP POST mode it runs ONE sendPostHandler goroutine which calls handleSendPostMessage synchronously, so the whole process has at most one RPC in flight. One stuck call therefore blocks the block ingestor, the mempool monitor and every user-facing gRPC handler at once. These compound: a timeout alone would not have been enough, because bounding the caller's wait still leaves the shared goroutine stuck inside http.Client.Do with everything queued behind it. Only a timeout on the HTTP client aborts the in-flight request, and only dropping the shared goroutine lets independent callers proceed. Patching the vendored copy is not an option here: go.mod declares go 1.12, so automatic vendor mode (go >= 1.14) is off and the committed vendor/ tree is silently ignored in favour of the module cache. It is also stale -- vendor/modules.txt disagrees with go.mod on btcd, protobuf, logrus, sqlite3 and six other modules, so -mod=vendor cannot build at all, and `go mod vendor` would discard any patch. Replace it with package zrpc: a ~150-line JSON-RPC client that is safe for concurrent use and takes a timeout. The request envelope, ID sequence, basic auth and error semantics are deliberately identical. RPCError.Error() still renders as ": " because callers recover the numeric code from the string -- common.GetSaplingInfo checks for -8 via strings.SplitN(err.Error(), ":", 2) -- and the non-JSON body path still reports `status code: %d, response: %q`. Default timeout 120s, tunable with -rpc-timeout (0 disables). The bound is set by the slowest legitimate call: coinsupply measured 48s against a cold UTXO set, 3s once cached, so a tighter timeout would turn a slow-but-working call into a hard failure. Also drops rpcclient's Close=true, which opened a fresh TCP connection per request and left hundreds of sockets in TIME_WAIT on a busy node; idle connections are now reused and capped. Verified against a live dragonxd: getblockchaininfo returns chain=main; a bad height yields exactly "-8: Block height out of range" and parses back to -8; a 1ns timeout aborts in 54us instead of hanging; and the built daemon serves GetLightdInfo at the current height with no errors. The pre-existing parser TestCompactBlocks failure is unrelated and reproduces on the base commit. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FU87LdsJZiZkfq1eXubpeo --- cmd/server/main.go | 7 +- common/common.go | 14 ++-- frontend/mempool.go | 6 +- frontend/rpc_client.go | 32 ++++----- frontend/service.go | 6 +- zrpc/client.go | 149 +++++++++++++++++++++++++++++++++++++++++ zrpc/live_test.go | 86 ++++++++++++++++++++++++ 7 files changed, 270 insertions(+), 30 deletions(-) create mode 100644 zrpc/client.go create mode 100644 zrpc/live_test.go 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) + } +}