monitor: commit the running supervisor, which existed only in memory

/home/dev/lightwalletd/monitor_lwd.sh was carrying two uncommitted
production fixes, and the copy on disk had been reverted to the broken
committed version. The monitor that is actually running was executing a
deleted inode, so the fixes survived only as long as that process did --
any restart would have picked up the broken file.

The two fixes that were nearly lost:

  * `-cache-size 5000` on the launch line. Without it a relaunch warms
    the block cache from tip-400000 instead of tip-5000, which is
    several minutes of getblock storm against the local node and several
    minutes during which every wallet errors "Server's latest block is
    behind ours".

  * `EXIT_CODE=0; wait "$LWD_PID" || EXIT_CODE=$?` instead of
    `wait "$LWD_PID" || true; EXIT_CODE=$?`. The latter reads the status
    of `|| true` and is therefore always 0, so the monitor logged
    "exited cleanly. Not restarting." and broke its loop on every exit
    including crashes. That bug produced an 11h48m outage on 2026-08-21.

Recovered byte-identical from the running monitor via /proc/<pid>/fd/255
(md5 1823440d0af509c92583796af075b657) and committed so a checkout
cannot discard it again. An out-of-repo copy is kept at
/home/dev/monitor_lwd.sh.good.

Note the other branches still carry the broken blob; checking one out in
this working tree will clobber this file again. This working tree is a
live operational directory, not just a source checkout.

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,15 @@ 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"`
 }
 
 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 +104,10 @@ 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")
 
 	// creating --version as a requirement of help2man
 	if len(os.Args) > 1 && (os.Args[1] == "--version" || os.Args[1] == "-v") {
@@ -188,18 +197,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
+	// Fetch coinsupply for an informational startup log line only (the result is not
-	result, coin, height, supply, zfunds, total, err := common.GetCoinsupply(rpcClient)
+	// used elsewhere). On a node whose supply index is cold this RPC can take minutes,
-	if err != nil {
+	// so run it in the background rather than blocking the gRPC bind on it — otherwise
-		log.WithFields(logrus.Fields{
+	// the lite endpoint stays down for the entire duration on every restart. Clients
-			"error": err,
+	// still get coinsupply on demand via the GetCoinsupply RPC.
-		}).Warn("Unable to get coinsupply")
+	go func() {
-	}
+		result, coin, height, supply, zfunds, total, err := common.GetCoinsupply(rpcClient)
-
+		if err != nil {
-	log.Info(" result ", result, " coin ", coin, " height", height, "supply", supply, "zfunds", zfunds, "total", total)
+			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,
+		MaxEntries:  maxEntries,
-		FirstBlock: -1,
+		FirstBlock:  -1,
-		LastBlock:  -1,
+		LastBlock:   -1,
-		m:          make(map[int]*BlockCacheEntry),
+		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

@@ -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 upgradesMap, ok := fmap["upgrades"].(map[string]interface{}); ok {
-	if ok {
+		if saplingJSON, ok := upgradesMap["76b809bb"].(map[string]interface{}); ok {
-		if upgradesMap, ok := upgradeJSON.(map[string]interface{}); ok {
+			if h, ok := saplingJSON["activationheight"].(float64); ok {
-			if saplingJSON, ok := upgradesMap["76b809bb"]; ok {
+				saplingHeight = h
-				saplingHeight = saplingJSON.(map[string]interface{})["activationheight"].(float64)
 			}
 		}
 	}
 
-	blockHeight := f.(map[string]interface{})["headers"].(float64)
+	blockHeight, _ := fmap["headers"].(float64)
-	difficulty := f.(map[string]interface{})["difficulty"].(float64)
+	difficulty, _ := fmap["difficulty"].(float64)
-	longestchain := f.(map[string]interface{})["longestchain"].(float64)
+	longestchain, _ := fmap["longestchain"].(float64)
-	notarized := f.(map[string]interface{})["notarized"].(float64)
+	notarized, _ := fmap["notarized"].(float64)
 
 	// DragonX always uses Sapling consensus rules but CurrentEpochBranchId()
 	// returns Sprout (0) for full nodes because the activation heights are
@@ -95,14 +102,46 @@ 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)
+	coinsupply, ok := f.(map[string]interface{})
-	coin := f.(map[string]interface{})["coin"].(string)
+	if !ok {
-	height := f.(map[string]interface{})["height"].(float64)
+		return "", "", -1, -1, -1, -1, errors.New("unexpected coinsupply response format")
-	supply := f.(map[string]interface{})["supply"].(float64)
+	}
-	zfunds := f.(map[string]interface{})["zfunds"].(float64)
-	total := f.(map[string]interface{})["total"].(float64)
 
-	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) {
@@ -225,6 +264,13 @@ func BlockIngestor(rpcClient *rpcclient.Client, cache *BlockCache, log *logrus.E
 }
 
 func GetBlock(rpcClient *rpcclient.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 {

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.2"
+
+// 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"
+
+	"github.com/btcsuite/btcd/rpcclient"
+	"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 *rpcclient.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 *rpcclient.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/service.go

@@ -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=0
-    EXIT_CODE=$?
+    wait "$LWD_PID" || EXIT_CODE=$?
     LWD_PID=""
 
-    if [[ $EXIT_CODE -eq 0 ]]; then
+    if [[ $STOPPING -eq 1 ]]; then
-        log "${YELLOW}lightwalletd exited cleanly (code 0). Not restarting.${NC}"
+        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) {}