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 "<code>: <message>" 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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FU87LdsJZiZkfq1eXubpeo

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{

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

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

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
 }

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
 }
 

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 "<code>: <message>", 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 "<code>: <message>". 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
+}

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 "<code>: <message>" -- 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)
+	}
+}