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

@@ -85,10 +85,18 @@ type Options struct {
 	logPath     string `json:"log_file,omitempty"`
 	confPath    string `json:"conf_file,omitempty"`
 	cacheSize   int    `json:"cache_size,omitempty"`
+
+	adaptiveLag  bool `json:"adaptive_lag,omitempty"`
+	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() {
-	var version = "0.1.1" // set version number
+	var version = common.Version
 
 	opts := &Options{}
 	flag.StringVar(&opts.bindAddr, "bind-addr", "127.0.0.1:9069", "the address to listen on")
@@ -99,6 +107,12 @@ func main() {
 	flag.StringVar(&opts.logPath, "log-file", "", "log file to write to")
 	flag.StringVar(&opts.confPath, "conf-file", "", "conf file to pull RPC creds from")
 	flag.IntVar(&opts.cacheSize, "cache-size", 400000, "number of blocks to hold in the cache")
+	flag.BoolVar(&opts.adaptiveLag, "adaptive-lag", true, "advertise a tip that trails the real tip by a reorg-rate-driven lag, so wallets anchor shielded spends at a settled height")
+	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") {
@@ -110,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)
@@ -163,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{
@@ -188,18 +209,25 @@ func main() {
 
 	log.Info("Got sapling height ", saplingHeight, " chain ", chainName, " branchID ", branchID, " difficulty ", difficulty, longestchain, " longestchain ", notarized, " notarized ")
 
-	// Get the Coinsupply from the RPC
-	result, coin, height, supply, zfunds, total, err := common.GetCoinsupply(rpcClient)
-	if err != nil {
-		log.WithFields(logrus.Fields{
-			"error": err,
-		}).Warn("Unable to get coinsupply")
-	}
-
-	log.Info(" result ", result, " coin ", coin, " height", height, "supply", supply, "zfunds", zfunds, "total", total)
+	// Fetch coinsupply for an informational startup log line only (the result is not
+	// used elsewhere). On a node whose supply index is cold this RPC can take minutes,
+	// so run it in the background rather than blocking the gRPC bind on it — otherwise
+	// the lite endpoint stays down for the entire duration on every restart. Clients
+	// still get coinsupply on demand via the GetCoinsupply RPC.
+	go func() {
+		result, coin, height, supply, zfunds, total, err := common.GetCoinsupply(rpcClient)
+		if err != nil {
+			log.WithFields(logrus.Fields{
+				"error": err,
+			}).Warn("Unable to get coinsupply")
+			return
+		}
+		log.Info(" result ", result, " coin ", coin, " height", height, "supply", supply, "zfunds", zfunds, "total", total)
+	}()
 
 	// Initialize the cache
 	cache := common.NewBlockCache(opts.cacheSize)
+	cache.ConfigureLag(opts.adaptiveLag, opts.lagWindowMin, opts.lagMin, opts.lagMax)
 
 	stopChan := make(chan bool, 1)
 

common/cache.go

@@ -3,6 +3,7 @@ package common
 import (
 	"bytes"
 	"sync"
+	"time"
 
 	"git.hush.is/hush/lightwalletd/walletrpc"
 	"github.com/golang/protobuf/proto"
@@ -22,14 +23,34 @@ type BlockCache struct {
 	m map[int]*BlockCacheEntry
 
 	mutex sync.RWMutex
+
+	// Adaptive confirmation lag. lightwalletd advertises a tip that trails the
+	// real tip by a number of blocks that scales with the recent reorg rate, so
+	// wallets anchor shielded spends at a settled height while the chain is
+	// churning, and at (near) the real tip when it is stable. reorgTimes holds
+	// the times of recent reorg detections, pruned to lagWindow.
+	reorgTimes  []time.Time
+	lagWindow   time.Duration
+	lagMin      int
+	lagMax      int
+	adaptiveLag bool
+
+	// advTip is the cached advertised (lag-adjusted) tip, kept fresh by Add so
+	// the per-block serving cap (HeightAllowed) is a cheap read. -1 until ready.
+	advTip int
 }
 
 func NewBlockCache(maxEntries int) *BlockCache {
 	return &BlockCache{
-		MaxEntries: maxEntries,
-		FirstBlock: -1,
-		LastBlock:  -1,
-		m:          make(map[int]*BlockCacheEntry),
+		MaxEntries:  maxEntries,
+		FirstBlock:  -1,
+		LastBlock:   -1,
+		m:           make(map[int]*BlockCacheEntry),
+		adaptiveLag: true,
+		lagWindow:   30 * time.Minute,
+		lagMin:      1,
+		lagMax:      12,
+		advTip:      -1,
 	}
 }
 
@@ -57,6 +78,8 @@ func (c *BlockCache) Add(height int, block *walletrpc.CompactBlock) (error, bool
 	// Don't allow out-of-order blocks. This is more of a sanity check than anything
 	// If there is a reorg, then the ingestor needs to handle it.
 	if c.m[height-1] != nil && !bytes.Equal(block.PrevHash, c.m[height-1].hash) {
+		// Record the reorg so the adaptive confirmation lag can react to it.
+		c.reorgTimes = append(c.reorgTimes, time.Now())
 		return nil, true
 	}
 
@@ -81,6 +104,12 @@ func (c *BlockCache) Add(height int, block *walletrpc.CompactBlock) (error, bool
 		c.FirstBlock = c.FirstBlock + 1
 	}
 
+	// Keep the advertised (lag-adjusted) tip fresh for the block-serving cap.
+	c.advTip = c.LastBlock - c.computeLagLocked()
+	if c.advTip < c.FirstBlock {
+		c.advTip = c.FirstBlock
+	}
+
 	//println("Cache size is ", len(c.m))
 	return nil, false
 }
@@ -116,3 +145,78 @@ func (c *BlockCache) GetLatestBlock() int {
 
 	return c.LastBlock
 }
+
+// ConfigureLag sets the adaptive-confirmation-lag parameters. Called once at
+// startup from the command-line flags.
+func (c *BlockCache) ConfigureLag(adaptive bool, windowMinutes, min, max int) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	c.adaptiveLag = adaptive
+	c.lagWindow = time.Duration(windowMinutes) * time.Minute
+	c.lagMin = min
+	c.lagMax = max
+}
+
+// computeLagLocked returns the confirmation lag (in blocks) to apply right now
+// and prunes reorg events that have aged out of the window. The lag is the
+// configured floor plus the number of reorgs seen within lagWindow, capped at
+// lagMax. Caller must hold c.mutex.
+func (c *BlockCache) computeLagLocked() int {
+	if !c.adaptiveLag {
+		return c.lagMin
+	}
+	cutoff := time.Now().Add(-c.lagWindow)
+	kept := c.reorgTimes[:0]
+	for _, t := range c.reorgTimes {
+		if t.After(cutoff) {
+			kept = append(kept, t)
+		}
+	}
+	c.reorgTimes = kept
+	lag := c.lagMin + len(kept)
+	if lag > c.lagMax {
+		lag = c.lagMax
+	}
+	if lag < c.lagMin {
+		lag = c.lagMin
+	}
+	return lag
+}
+
+// CurrentLag returns the confirmation lag currently being applied.
+func (c *BlockCache) CurrentLag() int {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	return c.computeLagLocked()
+}
+
+// AdvertisedLatestBlock is the tip height lightwalletd reports to wallets: the
+// real cache tip minus the adaptive confirmation lag. Wallets sync to and
+// anchor shielded spends at this settled height, immune to tip reorgs. The
+// internal cache and ingestor continue to track the real tip.
+func (c *BlockCache) AdvertisedLatestBlock() int {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	if c.LastBlock < 0 {
+		return c.LastBlock
+	}
+	adv := c.LastBlock - c.computeLagLocked()
+	if adv < c.FirstBlock {
+		adv = c.FirstBlock
+	}
+	c.advTip = adv
+	return adv
+}
+
+// HeightAllowed reports whether height is at or below the advertised
+// (lag-adjusted) tip. lightwalletd refuses to serve blocks above it so wallets
+// cannot sync or anchor shielded spends into the unstable reorg zone. Cheap
+// (read lock, no recompute); advTip is kept fresh by Add.
+func (c *BlockCache) HeightAllowed(height int) bool {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+	if c.advTip < 0 {
+		return true
+	}
+	return height <= c.advTip
+}

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
@@ -38,25 +38,32 @@ func GetSaplingInfo(rpcClient *rpcclient.Client) (int, int, string, string, int,
 		return -1, -1, "", "", -1, -1, -1, errors.Wrap(err, "error reading JSON response")
 	}
 
-	chainName := f.(map[string]interface{})["chain"].(string)
+	// Assert the top-level object once, then read every field with the comma-ok
+	// form. These run on the block-ingestor goroutine, and nothing in this
+	// daemon or in grpc-go v1.24.0 recovers a panic, so one unexpected response
+	// shape would take the whole process down rather than fail one call.
+	fmap, ok := f.(map[string]interface{})
+	if !ok {
+		return -1, -1, "", "", -1, -1, -1, errors.New("getblockchaininfo: unexpected response shape")
+	}
+	chainName, _ := fmap["chain"].(string)
 
 	// DragonX has Sapling active from block 1 but sets NO_ACTIVATION_HEIGHT in
 	// chainparams, so dragonxd omits it from the upgrades map. Fall back to
 	// height 1 when the key is absent.
 	saplingHeight := float64(1)
-	upgradeJSON, ok := f.(map[string]interface{})["upgrades"]
-	if ok {
-		if upgradesMap, ok := upgradeJSON.(map[string]interface{}); ok {
-			if saplingJSON, ok := upgradesMap["76b809bb"]; ok {
-				saplingHeight = saplingJSON.(map[string]interface{})["activationheight"].(float64)
+	if upgradesMap, ok := fmap["upgrades"].(map[string]interface{}); ok {
+		if saplingJSON, ok := upgradesMap["76b809bb"].(map[string]interface{}); ok {
+			if h, ok := saplingJSON["activationheight"].(float64); ok {
+				saplingHeight = h
 			}
 		}
 	}
 
-	blockHeight := f.(map[string]interface{})["headers"].(float64)
-	difficulty := f.(map[string]interface{})["difficulty"].(float64)
-	longestchain := f.(map[string]interface{})["longestchain"].(float64)
-	notarized := f.(map[string]interface{})["notarized"].(float64)
+	blockHeight, _ := fmap["headers"].(float64)
+	difficulty, _ := fmap["difficulty"].(float64)
+	longestchain, _ := fmap["longestchain"].(float64)
+	notarized, _ := fmap["notarized"].(float64)
 
 	// DragonX always uses Sapling consensus rules but CurrentEpochBranchId()
 	// returns Sprout (0) for full nodes because the activation heights are
@@ -72,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
@@ -95,17 +102,49 @@ func GetCoinsupply(rpcClient *rpcclient.Client) (string, string, int, int, int,
 		return "", "", -1, -1, -1, -1, errors.Wrap(err, "error reading JSON response")
 	}
 
-	result := f.(map[string]interface{})["result"].(string)
-	coin := f.(map[string]interface{})["coin"].(string)
-	height := f.(map[string]interface{})["height"].(float64)
-	supply := f.(map[string]interface{})["supply"].(float64)
-	zfunds := f.(map[string]interface{})["zfunds"].(float64)
-	total := f.(map[string]interface{})["total"].(float64)
+	coinsupply, ok := f.(map[string]interface{})
+	if !ok {
+		return "", "", -1, -1, -1, -1, errors.New("unexpected coinsupply response format")
+	}
 
-	return result, coin, int(height), int(supply), int(zfunds), int(total), nil
+	getStringField := func(key string) string {
+		value, ok := coinsupply[key]
+		if !ok || value == nil {
+			return ""
+		}
+
+		if strValue, ok := value.(string); ok {
+			return strValue
+		}
+
+		return fmt.Sprintf("%v", value)
+	}
+
+	getNumberField := func(key string) int {
+		value, ok := coinsupply[key]
+		if !ok || value == nil {
+			return 0
+		}
+
+		number, ok := value.(float64)
+		if !ok {
+			return 0
+		}
+
+		return int(number)
+	}
+
+	result := getStringField("result")
+	coin := getStringField("coin")
+	height := getNumberField("height")
+	supply := getNumberField("supply")
+	zfunds := getNumberField("zfunds")
+	total := getNumberField("total")
+
+	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")
@@ -148,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
@@ -224,7 +263,14 @@ 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) {
+		return nil, errors.New(
+			fmt.Sprintf("Block %d is above the advertised tip (reorg-lag protection)", height))
+	}
+
 	// First, check the cache to see if we have the block
 	block := cache.Get(height)
 	if block != nil {
@@ -247,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

@@ -0,0 +1,14 @@
+package common
+
+// Version is the single source of truth for this daemon's version.
+//
+// It was previously duplicated as a literal in two places that could drift:
+// cmd/server/main.go's --version output and the Version field of the LightdInfo
+// gRPC reply in frontend/service.go. The gRPC one is the load-bearing copy --
+// 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.3"
+
+// VersionString is what GetLightdInfo advertises to clients.
+const VersionString = Version + "-dragonxlightd"

frontend/mempool.go

@@ -0,0 +1,173 @@
+package frontend
+
+import (
+	"encoding/hex"
+	"encoding/json"
+	"sync"
+	"time"
+
+	"git.hush.is/hush/lightwalletd/zrpc"
+	"github.com/sirupsen/logrus"
+
+	"git.hush.is/hush/lightwalletd/common"
+	"git.hush.is/hush/lightwalletd/walletrpc"
+)
+
+// mempoolMonitor polls the node's mempool ONCE (regardless of how many wallets are streaming) and
+// fans each new transaction out to every subscribed GetMempoolStream handler. This replaces
+// per-connection polling: N connected wallets no longer each hit getrawmempool/getrawtransaction.
+//
+// Lifecycle matches the semantics the client's monitor loop expects: while a block is current the
+// monitor emits each mempool tx to subscribers exactly once; when a new block is mined it resets
+// and closes all subscriber channels, so each handler returns (closing its stream) and the client
+// reconnects after re-syncing the block. The poller starts lazily on the first subscription and,
+// when no clients are connected, idles without touching the node.
+type mempoolMonitor struct {
+	mu      sync.Mutex
+	subs    map[int]chan *walletrpc.RawTransaction
+	nextID  int
+	seen    map[string]bool             // txids already emitted for the current block
+	ordered []*walletrpc.RawTransaction // emitted txs in arrival order, replayed to late subscribers
+}
+
+var (
+	sharedMempoolMonitor *mempoolMonitor
+	mempoolMonitorOnce   sync.Once
+)
+
+// 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 *zrpc.Client, cache *common.BlockCache, log *logrus.Entry) *mempoolMonitor {
+	mempoolMonitorOnce.Do(func() {
+		sharedMempoolMonitor = &mempoolMonitor{
+			subs: make(map[int]chan *walletrpc.RawTransaction),
+			seen: make(map[string]bool),
+		}
+		go sharedMempoolMonitor.run(client, cache, log)
+	})
+	return sharedMempoolMonitor
+}
+
+// subscribe registers a stream. It returns the subscriber id, a channel of subsequent mempool txs,
+// and a snapshot of txs already emitted this block (to be sent first). Registration and snapshot
+// are taken atomically, so every tx reaches a subscriber exactly once (snapshot xor channel).
+func (m *mempoolMonitor) subscribe() (int, chan *walletrpc.RawTransaction, []*walletrpc.RawTransaction) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	id := m.nextID
+	m.nextID++
+	ch := make(chan *walletrpc.RawTransaction, 256)
+	m.subs[id] = ch
+	snapshot := make([]*walletrpc.RawTransaction, len(m.ordered))
+	copy(snapshot, m.ordered)
+	return id, ch, snapshot
+}
+
+// unsubscribe removes a stream. Safe to call after a new-block reset already dropped it (the map
+// lookup guards against a double close).
+func (m *mempoolMonitor) unsubscribe(id int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if ch, ok := m.subs[id]; ok {
+		delete(m.subs, id)
+		close(ch)
+	}
+}
+
+// reset is called when a new block is mined: forget this block's txs and close every subscriber
+// stream so clients re-sync the block and reconnect.
+func (m *mempoolMonitor) reset() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.seen = make(map[string]bool)
+	m.ordered = nil
+	for id, ch := range m.subs {
+		delete(m.subs, id)
+		close(ch)
+	}
+}
+
+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
+		// height that advanced during the idle period.
+		if h := cache.GetLatestBlock(); h != lastHeight {
+			lastHeight = h
+			m.reset()
+		}
+
+		m.mu.Lock()
+		idle := len(m.subs) == 0
+		m.mu.Unlock()
+		if idle {
+			// No one is listening; don't touch the node until a client subscribes.
+			time.Sleep(2 * time.Second)
+			continue
+		}
+
+		height := uint64(lastHeight + 1)
+
+		// List the current mempool txids: getrawmempool(false) -> ["txid", ...].
+		mpParams := []json.RawMessage{json.RawMessage("false")}
+		result, rpcErr := client.RawRequest("getrawmempool", mpParams)
+		if rpcErr != nil {
+			log.Warningf("mempool monitor: getrawmempool failed: %s", rpcErr.Error())
+			time.Sleep(2 * time.Second)
+			continue
+		}
+		var txids []string
+		if err := json.Unmarshal(result, &txids); err != nil {
+			log.Warningf("mempool monitor: cannot parse getrawmempool: %s", err.Error())
+			time.Sleep(2 * time.Second)
+			continue
+		}
+
+		for _, txid := range txids {
+			m.mu.Lock()
+			already := m.seen[txid]
+			m.mu.Unlock()
+			if already {
+				continue
+			}
+
+			// Fetch the full serialized tx: getrawtransaction("txid") -> hex. Full bytes are
+			// required so shielded memos survive (a CompactTx's 52-byte prefix would not carry a
+			// memo). The mempool txid is already in RPC display byte order, so (unlike
+			// GetTransaction) no reversal is needed.
+			txParams := []json.RawMessage{json.RawMessage("\"" + txid + "\"")}
+			txResult, txErr := client.RawRequest("getrawtransaction", txParams)
+			if txErr != nil {
+				// The tx may have been mined or evicted between listing and fetch; skip it.
+				continue
+			}
+			var txhex string
+			if err := json.Unmarshal(txResult, &txhex); err != nil {
+				continue
+			}
+			txBytes, err := hex.DecodeString(txhex)
+			if err != nil {
+				continue
+			}
+			rtx := &walletrpc.RawTransaction{Data: txBytes, Height: height}
+
+			// Record + fan out under the lock. The send is non-blocking (default case), so a slow
+			// subscriber never stalls the poller or the other subscribers.
+			m.mu.Lock()
+			if !m.seen[txid] {
+				m.seen[txid] = true
+				m.ordered = append(m.ordered, rtx)
+				for _, ch := range m.subs {
+					select {
+					case ch <- rtx:
+					default:
+						// Slow subscriber; drop. Block sync will deliver the confirmed tx later.
+					}
+				}
+			}
+			m.mu.Unlock()
+		}
+
+		time.Sleep(2 * time.Second)
+	}
+}

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
 }
 
@@ -41,9 +41,12 @@ func (s *SqlStreamer) GetCache() *common.BlockCache {
 }
 
 func (s *SqlStreamer) GetLatestBlock(ctx context.Context, placeholder *walletrpc.ChainSpec) (*walletrpc.BlockID, error) {
-	latestBlock := s.cache.GetLatestBlock()
+	// Advertise a tip that trails the real tip by the adaptive confirmation lag,
+	// so wallets anchor shielded spends at a settled height during reorg churn.
+	latestBlock := s.cache.AdvertisedLatestBlock()
 	s.log.WithFields(logrus.Fields{
 		"latestBlock": latestBlock,
+		"adaptiveLag": s.cache.CurrentLag(),
 	}).Info("GetLatestBlock called")
 
 	if latestBlock == -1 {
@@ -116,6 +119,45 @@ func (s *SqlStreamer) GetAddressTxids(addressBlockFilter *walletrpc.TransparentA
 	return nil
 }
 
+// GetMempoolStream streams currently-unconfirmed (mempool) transactions to the client as full
+// RawTransactions, so lightwallets can detect 0-confirmation transactions -- including reading
+// shielded memos (e.g. incoming chat messages) -- without waiting for a block. Each mempool tx is
+// emitted once; the stream stays open until a new block is mined, at which point it returns
+// (closing the stream) so the client re-syncs the block and reconnects. This mirrors the
+// zecwallet/Hush lightwalletd semantics the wallets already expect.
+//
+// The node's mempool is polled by a single process-wide monitor that fans out to all subscribers
+// (see mempool.go), so the node isn't polled once per connected wallet.
+func (s *SqlStreamer) GetMempoolStream(_ *walletrpc.Empty, resp walletrpc.CompactTxStreamer_GetMempoolStreamServer) error {
+	monitor := getMempoolMonitor(s.client, s.cache, s.log)
+	id, ch, snapshot := monitor.subscribe()
+	defer monitor.unsubscribe(id)
+
+	// First send the txs already in this block's mempool, so a wallet that connects mid-block still
+	// sees them.
+	for _, rtx := range snapshot {
+		if err := resp.Send(rtx); err != nil {
+			return err
+		}
+	}
+
+	// Then stream subsequent txs until the monitor closes the channel (a new block was mined) or
+	// the client disconnects.
+	for {
+		select {
+		case <-resp.Context().Done():
+			return nil
+		case rtx, ok := <-ch:
+			if !ok {
+				return nil
+			}
+			if err := resp.Send(rtx); err != nil {
+				return err
+			}
+		}
+	}
+}
+
 func (s *SqlStreamer) GetBlock(ctx context.Context, id *walletrpc.BlockID) (*walletrpc.CompactBlock, error) {
 	if id.Height == 0 && id.Hash == nil {
 		return nil, ErrUnspecified
@@ -251,7 +293,26 @@ func (s *SqlStreamer) GetTransaction(ctx context.Context, txf *walletrpc.TxFilte
 		if err != nil {
 			return nil, err
 		}
-		txHeight = txinfo.(map[string]interface{})["height"].(float64)
+		// dragonxd emits "height" only for a transaction that is in a block:
+		// rawtransaction.cpp puts it inside `if (!hashBlock.IsNull())`. Every
+		// mempool transaction therefore comes back WITHOUT the key, and an
+		// unchecked assertion on the missing value panics -- which, with no
+		// recover() anywhere in grpc-go v1.24.0 or in this daemon, kills the
+		// whole process and every wallet connected to it. Any client can reach
+		// this by broadcasting a transaction and immediately asking for it, and
+		// GetMempoolStream hands out unconfirmed txids by design.
+		//
+		// Report an unconfirmed transaction as tip+1, matching what
+		// GetMempoolStream already advertises (mempool.go:109).
+		txmap, ok := txinfo.(map[string]interface{})
+		if !ok {
+			return nil, errors.New("getrawtransaction: unexpected response shape")
+		}
+		if h, ok := txmap["height"].(float64); ok {
+			txHeight = h
+		} else {
+			txHeight = float64(s.cache.GetLatestBlock() + 1)
+		}
 
 		return &walletrpc.RawTransaction{Data: txBytes, Height: uint64(txHeight)}, nil
 	}
@@ -282,16 +343,23 @@ func (s *SqlStreamer) GetLightdInfo(ctx context.Context, in *walletrpc.Empty) (*
 		return nil, err
 	}
 
+	// Report the advertised (lag-adjusted) tip so wallets consider themselves
+	// synced at the height they are actually served, not perpetually N behind.
+	advHeight := s.cache.AdvertisedLatestBlock()
+	if advHeight < 0 {
+		advHeight = blockHeight
+	}
+
 	// TODO these are called Error but they aren't at the moment.
 	// A success will return code 0 and message txhash.
 	return &walletrpc.LightdInfo{
-		Version:                 "0.1.1-dragonxlightd",
+		Version:                 common.VersionString,
 		Vendor:                  "DragonX LightWalletD",
 		TaddrSupport:            true,
 		ChainName:               chainName,
 		SaplingActivationHeight: uint64(saplingHeight),
 		ConsensusBranchId:       consensusBranchId,
-		BlockHeight:             uint64(blockHeight),
+		BlockHeight:             uint64(advHeight),
 		Difficulty:              uint64(difficulty),
 		Longestchain:            uint64(longestchain),
 		Notarized:               uint64(notarized),

monitor_lwd.sh

@@ -10,9 +10,10 @@ set -euo pipefail
 
 SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
 LWD_BIN="$SCRIPT_DIR/lightwalletd"
-LWD_ARGS="-bind-addr lite.dragonx.is:9069 -conf-file $HOME/.hush/DRAGONX/DRAGONX.conf -no-tls"
+LWD_ARGS="-bind-addr lite.dragonx.is:9069 -conf-file $HOME/.hush/DRAGONX/DRAGONX.conf -no-tls -lag-min 4 -lag-max 12 -lag-window 30 -cache-size 5000"
 LOGFILE="$SCRIPT_DIR/lwd-monitor.log"
 PIDFILE="/tmp/lwd-monitor.pid"
+STOPPING=0
 RESTART_DELAY=5        # seconds to wait before restarting after a crash
 MAX_RAPID_RESTARTS=5   # max restarts within the rapid window before backing off
 RAPID_WINDOW=120       # seconds — if this many restarts happen within this window, back off
@@ -29,6 +30,7 @@ log() {
 }
 
 cleanup() {
+    STOPPING=1
     log "${YELLOW}Monitor shutting down...${NC}"
     if [[ -n "${LWD_PID:-}" ]] && kill -0 "$LWD_PID" 2>/dev/null; then
         log "Stopping lightwalletd (PID $LWD_PID)..."
@@ -75,6 +77,7 @@ log "Args:    $LWD_ARGS"
 
 restart_times=()
 LWD_PID=""
+STOPPING=0
 
 while true; do
     # Start lightwalletd
@@ -84,12 +87,12 @@ while true; do
     log "lightwalletd started with PID $LWD_PID"
 
     # Wait for it to exit
-    wait "$LWD_PID" || true
-    EXIT_CODE=$?
+    EXIT_CODE=0
+    wait "$LWD_PID" || EXIT_CODE=$?
     LWD_PID=""
 
-    if [[ $EXIT_CODE -eq 0 ]]; then
-        log "${YELLOW}lightwalletd exited cleanly (code 0). Not restarting.${NC}"
+    if [[ $STOPPING -eq 1 ]]; then
+        log "${YELLOW}lightwalletd stopped on request (code $EXIT_CODE). Not restarting.${NC}"
         break
     fi
 

walletrpc/service.pb.go

@@ -694,6 +694,8 @@ type CompactTxStreamerClient interface {
 	SendTransaction(ctx context.Context, in *RawTransaction, opts ...grpc.CallOption) (*SendResponse, error)
 	// t-Address support
 	GetAddressTxids(ctx context.Context, in *TransparentAddressBlockFilter, opts ...grpc.CallOption) (CompactTxStreamer_GetAddressTxidsClient, error)
+	// Mempool
+	GetMempoolStream(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_GetMempoolStreamClient, error)
 	// Misc
 	GetLightdInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*LightdInfo, error)
 	GetCoinsupply(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Coinsupply, error)
@@ -807,6 +809,38 @@ func (x *compactTxStreamerGetAddressTxidsClient) Recv() (*RawTransaction, error)
 	return m, nil
 }
 
+func (c *compactTxStreamerClient) GetMempoolStream(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_GetMempoolStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &_CompactTxStreamer_serviceDesc.Streams[2], "/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetMempoolStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &compactTxStreamerGetMempoolStreamClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type CompactTxStreamer_GetMempoolStreamClient interface {
+	Recv() (*RawTransaction, error)
+	grpc.ClientStream
+}
+
+type compactTxStreamerGetMempoolStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *compactTxStreamerGetMempoolStreamClient) Recv() (*RawTransaction, error) {
+	m := new(RawTransaction)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 func (c *compactTxStreamerClient) GetLightdInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*LightdInfo, error) {
 	out := new(LightdInfo)
 	err := c.cc.Invoke(ctx, "/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetLightdInfo", in, out, opts...)
@@ -835,6 +869,8 @@ type CompactTxStreamerServer interface {
 	SendTransaction(context.Context, *RawTransaction) (*SendResponse, error)
 	// t-Address support
 	GetAddressTxids(*TransparentAddressBlockFilter, CompactTxStreamer_GetAddressTxidsServer) error
+	// Mempool
+	GetMempoolStream(*Empty, CompactTxStreamer_GetMempoolStreamServer) error
 	// Misc
 	GetLightdInfo(context.Context, *Empty) (*LightdInfo, error)
 	GetCoinsupply(context.Context, *Empty) (*Coinsupply, error)
@@ -862,6 +898,9 @@ func (*UnimplementedCompactTxStreamerServer) SendTransaction(ctx context.Context
 func (*UnimplementedCompactTxStreamerServer) GetAddressTxids(req *TransparentAddressBlockFilter, srv CompactTxStreamer_GetAddressTxidsServer) error {
 	return status.Errorf(codes.Unimplemented, "method GetAddressTxids not implemented")
 }
+func (*UnimplementedCompactTxStreamerServer) GetMempoolStream(req *Empty, srv CompactTxStreamer_GetMempoolStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method GetMempoolStream not implemented")
+}
 func (*UnimplementedCompactTxStreamerServer) GetLightdInfo(ctx context.Context, req *Empty) (*LightdInfo, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetLightdInfo not implemented")
 }
@@ -987,6 +1026,27 @@ func (x *compactTxStreamerGetAddressTxidsServer) Send(m *RawTransaction) error {
 	return x.ServerStream.SendMsg(m)
 }
 
+func _CompactTxStreamer_GetMempoolStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(Empty)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(CompactTxStreamerServer).GetMempoolStream(m, &compactTxStreamerGetMempoolStreamServer{stream})
+}
+
+type CompactTxStreamer_GetMempoolStreamServer interface {
+	Send(*RawTransaction) error
+	grpc.ServerStream
+}
+
+type compactTxStreamerGetMempoolStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *compactTxStreamerGetMempoolStreamServer) Send(m *RawTransaction) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 func _CompactTxStreamer_GetLightdInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(Empty)
 	if err := dec(in); err != nil {
@@ -1063,6 +1123,11 @@ var _CompactTxStreamer_serviceDesc = grpc.ServiceDesc{
 			Handler:       _CompactTxStreamer_GetAddressTxids_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "GetMempoolStream",
+			Handler:       _CompactTxStreamer_GetMempoolStream_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "service.proto",
 }

walletrpc/service.proto

@@ -87,6 +87,13 @@ service CompactTxStreamer {
     // t-Address support
     rpc GetAddressTxids(TransparentAddressBlockFilter) returns (stream RawTransaction) {}
 
+    // Mempool
+    // Return a stream of current mempool transactions as full RawTransactions. The stream stays
+    // open while there are mempool transactions and is closed when a new block is mined, at which
+    // point the client re-syncs the block and reconnects. Full RawTransactions (not CompactTx) are
+    // required so wallets can read shielded memos of 0-confirmation transactions (e.g. chat).
+    rpc GetMempoolStream(Empty) returns (stream RawTransaction) {}
+
     // Misc
     rpc GetLightdInfo(Empty) returns (LightdInfo) {}
     rpc GetCoinsupply(Empty) returns (Coinsupply) {}

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)
+	}
+}