zrpc: bound concurrency, and close idle connections before the node does

Review follow-ups to 73a93f1.

Bound in-flight calls. rpcclient's single sendPostHandler goroutine
imposed an accidental ceiling of one concurrent RPC; removing it without
putting anything in its place left no bound at all. grpc-go supplies
none either -- this server sets no MaxConcurrentStreams, so the default
is math.MaxUint32 -- and dragonxd answers RPC with 8 worker threads
behind a 4096-deep queue, shared on the pool node with getblocktemplate.
Overload would therefore surface as mining latency rather than as an
error we could back off on. MaxConnsPerHost blocks the caller at the
limit instead of dialling more, which is the backpressure wanted;
MaxIdleConnsPerHost alone would only cap reuse and let us exceed the
limit while churning connections. Default 8, matching the node's
DEFAULT_HTTP_THREADS, tunable with -rpc-max-concurrent. Even 8 removes
all of the head-of-line blocking this work set out to fix.

IdleConnTimeout 90s -> 20s. dragonxd closes idle connections at 30s
(DEFAULT_HTTP_SERVER_TIMEOUT, applied via evhttp_set_timeout and not
overridden in DRAGONX.conf). At 90s we were always the second to close,
so a request could be written into a connection the server had already
sent a FIN for, and Go will not retry a POST once bytes are on the wire.
Closing first removes the race.

Reject a negative -rpc-timeout, which silently meant "unbounded", the
same as the documented 0. The check has to run after flag.Parse(); it
was initially placed before it and never fired.

Also correct the coinsupply note: hush_coinsupply walks the block index
back to genesis, loading each block from disk and memoising newcoins and
zfunds into the CBlockIndex, so the first call pays for the whole chain
and later ones are nearly free. It is not a UTXO-set scan, as the
earlier comment claimed. The measured 48s/3s figures are unchanged.

Verified: five concurrent GetLightdInfo calls all return grpc-status 0
with no errors, 46 blocks ingested, and the daemon holds 2 sockets to
the node rather than one per request; a negative timeout exits 1.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FU87LdsJZiZkfq1eXubpeo

cmd/server/main.go

@@ -91,7 +91,8 @@ type Options struct {
 	lagMax       int  `json:"lag_max,omitempty"`
 	lagWindowMin int  `json:"lag_window_min,omitempty"`
 
-	rpcTimeout time.Duration `json:"rpc_timeout,omitempty"`
+	rpcTimeout       time.Duration `json:"rpc_timeout,omitempty"`
+	rpcMaxConcurrent int           `json:"rpc_max_concurrent,omitempty"`
 }
 
 func main() {
@@ -110,7 +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. One stuck call otherwise blocks every other caller, because HTTP POST mode serialises all RPC through one goroutine")
+	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") {
@@ -122,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)
@@ -175,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, opts.rpcTimeout)
+	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", "", "", opts.rpcTimeout)
+		rpcClient, err = frontend.NewZRPCFromCreds("127.0.0.1:21769", "", "", opts.rpcTimeout, opts.rpcMaxConcurrent)
 
 		if err != nil {
 			log.WithFields(logrus.Fields{

frontend/rpc_client.go

@@ -12,14 +12,20 @@ import (
 // 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
+// makes is `coinsupply`, measured at 48s on first call and 3s afterwards.
-// node has cached the result). A timeout below that would turn a slow-but-
+// hush_coinsupply walks the block index back to genesis loading each block from
-// working call into a hard failure. 120s leaves ~2.5x headroom over the
+// disk, memoising newcoins/zfunds into the CBlockIndex as it goes, so the first
-// slowest real call while still bounding a hang that is otherwise unbounded --
+// call pays for the whole chain and later ones are nearly free. A timeout below
-// calls were seen running past five minutes before the process was killed.
+// 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
 
-func NewZRPCFromConf(confPath string, timeout time.Duration) (*zrpc.Client, error) {
+// 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")
@@ -30,10 +36,10 @@ func NewZRPCFromConf(confPath string, timeout time.Duration) (*zrpc.Client, erro
 	username := cfg.Section("").Key("rpcuser").String()
 	password := cfg.Section("").Key("rpcpassword").String()
 
-	return NewZRPCFromCreds(net.JoinHostPort(rpcaddr, rpcport), username, password, timeout)
+	return NewZRPCFromCreds(net.JoinHostPort(rpcaddr, rpcport), username, password, timeout, maxConcurrent)
 }
 
-func NewZRPCFromCreds(addr, username, password string, timeout time.Duration) (*zrpc.Client, error) {
+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), nil
+	return zrpc.New(addr, username, password, timeout, maxConcurrent), nil
 }

zrpc/client.go

@@ -71,10 +71,25 @@ type Client struct {
 	nextID int64
 }
 
-// New returns a client for a dragonxd JSON-RPC endpoint. A timeout of 0 means
+// New returns a client for a dragonxd JSON-RPC endpoint.
-// no timeout, which reproduces the old unbounded behaviour and should not be
+//
-// used in production.
+// timeout of 0 means no timeout, which reproduces the old unbounded behaviour
-func New(addr, user, pass string, timeout time.Duration) *Client {
+// 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,
@@ -87,9 +102,22 @@ func New(addr, user, pass string, timeout time.Duration) *Client {
 				// 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,
+				// MaxConnsPerHost is the real concurrency bound: it BLOCKS a
-				IdleConnTimeout:     90 * time.Second,
+				// 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,
 			},
 		},
 	}

zrpc/live_test.go

@@ -24,7 +24,7 @@ func liveClient(t *testing.T, timeout time.Duration) *Client {
 		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)
+	return New(k("rpcbind")+":"+k("rpcport"), k("rpcuser"), k("rpcpassword"), timeout, 8)
 }
 
 func TestLiveSuccess(t *testing.T) {
@@ -79,7 +79,7 @@ func TestLiveTimeoutFires(t *testing.T) {
 }
 
 func TestNoMethod(t *testing.T) {
-	c := New("127.0.0.1:1", "u", "p", time.Second)
+	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)
 	}