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
+	// 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,6 +23,21 @@ 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 {
@@ -30,6 +46,11 @@ func NewBlockCache(maxEntries int) *BlockCache {
 		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

@@ -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/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 {
@@ -321,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{
@@ -330,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