version: 0.1.3 for the zrpc transport change

Rebased onto master (v0.1.2, the deployed crash fix), so this branch now
carries only the RPC transport work. Bump so the deployed version stays
readable from off-box via GetLightdInfo -- the same check that verified
the 0.1.2 rollout node by node.

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,9 @@ 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"`
+	rpcMaxConcurrent int           `json:"rpc_max_concurrent,omitempty"`
 }
 
 func main() {
@@ -108,6 +111,8 @@ 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. Without it one stuck call blocks every other caller")
+	flag.IntVar(&opts.rpcMaxConcurrent, "rpc-max-concurrent", frontend.DefaultRPCMaxConcurrent, "maximum dragonxd JSON-RPC calls in flight at once; matches the node's RPC worker threads")
 
 	// creating --version as a requirement of help2man
 	if len(os.Args) > 1 && (os.Args[1] == "--version" || os.Args[1] == "-v") {
@@ -119,6 +124,13 @@ func main() {
 	// TODO support config from file and env vars
 	flag.Parse()
 
+	// A negative duration silently means "no timeout", the same as the
+	// documented 0, so reject it rather than quietly running unbounded.
+	if opts.rpcTimeout < 0 {
+		fmt.Fprintln(os.Stderr, "-rpc-timeout must not be negative; use 0 to disable the timeout")
+		os.Exit(1)
+	}
+
 	if opts.confPath == "" {
 		flag.Usage()
 		os.Exit(1)
@@ -172,13 +184,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, opts.rpcMaxConcurrent)
 	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, opts.rpcMaxConcurrent)
 
 		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

common/version.go

@@ -8,7 +8,7 @@ package common
 // it is walletrpc/service.proto:48, so every client and every operator probe
 // reads it, and it is the only way to tell from off-box which build a node is
 // running.
-const Version = "0.1.2"
+const Version = "0.1.3"
 
 // VersionString is what GetLightdInfo advertises to clients.
 const VersionString = Version + "-dragonxlightd"

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,30 @@ 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 on first call and 3s afterwards.
+// hush_coinsupply walks the block index back to genesis loading each block from
+// disk, memoising newcoins/zfunds into the CBlockIndex as it goes, so the first
+// call pays for the whole chain and later ones are nearly free. A timeout below
+// that first-call cost would turn a slow-but-working call into a hard failure.
+// 120s leaves ~2.5x headroom while still bounding a hang that is otherwise
+// unbounded -- calls were seen running past five minutes.
+const DefaultRPCTimeout = 120 * time.Second
+
+// DefaultRPCMaxConcurrent matches dragonxd's DEFAULT_HTTP_THREADS. Asking for
+// more in-flight calls than the node has worker threads only adds queueing.
+const DefaultRPCMaxConcurrent = 8
+
+func NewZRPCFromConf(confPath string, timeout time.Duration, maxConcurrent int) (*zrpc.Client, error) {
 	cfg, err := ini.Load(confPath)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to read config file")
@@ -19,19 +36,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, maxConcurrent)
 }
 
-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, maxConcurrent int) (*zrpc.Client, error) {
+	// DragonX only supports HTTP POST mode and does not provide TLS by default.
+	return zrpc.New(addr, username, password, timeout, maxConcurrent), 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,177 @@
+// 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.
+//
+// timeout of 0 means no timeout, which reproduces the old unbounded behaviour
+// and should not be used in production.
+//
+// maxConcurrent bounds how many requests may be in flight at once. This is not
+// optional book-keeping: rpcclient's single goroutine imposed an accidental
+// ceiling of ONE, and removing it without putting anything in its place would
+// let a burst of gRPC handlers fan out arbitrarily wide. grpc-go places no
+// limit of its own here -- this server sets no MaxConcurrentStreams, so the
+// default is math.MaxUint32. dragonxd serves RPC with 8 worker threads
+// (DEFAULT_HTTP_THREADS) behind a 4096-deep work queue, and on the pool node
+// those threads are shared with getblocktemplate, so overload shows up as
+// queueing latency for mining rather than as an error we could back off on.
+// A small number still removes all of the head-of-line blocking.
+func New(addr, user, pass string, timeout time.Duration, maxConcurrent int) *Client {
+	if maxConcurrent < 1 {
+		maxConcurrent = 1
+	}
+	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.
+				//
+				// MaxConnsPerHost is the real concurrency bound: it BLOCKS a
+				// caller once the limit is reached rather than dialling more,
+				// which is the backpressure we want. MaxIdleConnsPerHost only
+				// caps reuse, so on its own it would let us exceed the limit
+				// and go back to churning connections.
+				MaxConnsPerHost:     maxConcurrent,
+				MaxIdleConns:        maxConcurrent,
+				MaxIdleConnsPerHost: maxConcurrent,
+				// Must stay BELOW dragonxd's own idle timeout, which is 30s
+				// (DEFAULT_HTTP_SERVER_TIMEOUT in httpserver.h, applied via
+				// evhttp_set_timeout and not overridden in DRAGONX.conf).
+				// Whoever closes second loses a race against a FIN already in
+				// flight, and Go will not retry a POST once bytes are on the
+				// wire -- so we close first.
+				IdleConnTimeout: 20 * 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, 8)
+}
+
+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, 8)
+	if _, err := c.RawRequest("", nil); err == nil || err.Error() != "no method" {
+		t.Fatalf(`RawRequest("") = %v, want "no method"`, err)
+	}
+}