From 2bab58c6d23027e8045e6a3334a5da397dd531a4 Mon Sep 17 00:00:00 2001 From: DanS Date: Tue, 21 Jul 2026 01:05:10 -0500 Subject: [PATCH] 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 --- cmd/server/main.go | 34 ++++++++++---- common/cache.go | 112 ++++++++++++++++++++++++++++++++++++++++++-- common/common.go | 53 ++++++++++++++++++--- frontend/service.go | 14 +++++- monitor_lwd.sh | 2 +- 5 files changed, 192 insertions(+), 23 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index aa4c0ba..f766938 100644 --- a/cmd/server/main.go +++ b/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 - 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) diff --git a/common/cache.go b/common/cache.go index c2488a6..b328a57 100644 --- a/common/cache.go +++ b/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 +} diff --git a/common/common.go b/common/common.go index e063523..ba57fca 100644 --- a/common/common.go +++ b/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) - 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) { @@ -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 { diff --git a/frontend/service.go b/frontend/service.go index f8b66fc..280d9e4 100644 --- a/frontend/service.go +++ b/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), diff --git a/monitor_lwd.sh b/monitor_lwd.sh index 22ef60f..fdb8c6f 100755 --- a/monitor_lwd.sh +++ b/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