lightwalletd: adaptive reorg-lag tip + non-blocking startup coinsupply

Adaptive-lag: advertise a tip that trails the real tip by a reorg-rate-driven
lag (GetLatestBlock/GetLightdInfo) so wallets anchor shielded spends at a
settled height during reorg churn. Configurable via
-adaptive-lag/-lag-min/-lag-max/-lag-window; monitor_lwd.sh runs
-lag-min 4 -lag-max 12 -lag-window 30.

Startup coinsupply: the informational startup coinsupply RPC (result used only
for a log line) blocked the gRPC bind for minutes on a node whose supply index
is cold, keeping the lite endpoint down on every restart while it also starved
the block-cache ingestor. Run it in a goroutine so the bind is never blocked;
clients still get coinsupply on demand via the GetCoinsupply RPC.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

cmd/server/main.go

@@ -85,6 +85,11 @@ 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() {
@@ -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

@@ -95,14 +95,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 +257,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 {

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
@@ -282,6 +324,13 @@ 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{
@@ -291,7 +340,7 @@ func (s *SqlStreamer) GetLightdInfo(ctx context.Context, in *walletrpc.Empty) (*
 		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,7 +10,7 @@ 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"
 LOGFILE="$SCRIPT_DIR/lwd-monitor.log"
 PIDFILE="/tmp/lwd-monitor.pid"
 RESTART_DELAY=5        # seconds to wait before restarting after a crash

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