From be49b756adea09b1c204eb10929119d4a01d19df Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:05:03 +0200 Subject: [PATCH 01/13] bridge-monitor: on-chain balance fallback + degraded flag --- .../bridge-monitor/cmd/monitor/balance.go | 144 +++++++++++++++++- .../bridge-monitor/cmd/monitor/executor.go | 5 + .../bridge-monitor/cmd/monitor/metrics.go | 8 + .../cmd/monitor/onchain_balance.go | 22 +++ 4 files changed, 178 insertions(+), 1 deletion(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/balance.go b/harnesses/bridge-monitor/cmd/monitor/balance.go index 570a38fa..3d5c401b 100644 --- a/harnesses/bridge-monitor/cmd/monitor/balance.go +++ b/harnesses/bridge-monitor/cmd/monitor/balance.go @@ -9,6 +9,9 @@ import ( "os" "strings" "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/gagliardetto/solana-go" ) // BalanceChecker fetches wallet balances from Mobula API @@ -17,6 +20,18 @@ type BalanceChecker struct { apiKey string evmAddress string solanaAddress string + + // Fallback on-chain reader. When the Mobula portfolio API fails we read the + // triangle tokens directly from RPC instead of silently returning zeros, + // which used to make every tier look like "insufficient funds" during an + // API outage. Nil in quote-only mode (no TxExecutor). + onchain *TxExecutor +} + +// SetOnchainFallback wires the RPC-based balance readers. Called after the +// TxExecutor exists because the BalanceChecker is constructed first in main. +func (bc *BalanceChecker) SetOnchainFallback(tx *TxExecutor) { + bc.onchain = tx } // MobulaWalletResponse represents the Mobula wallet API response @@ -53,6 +68,17 @@ func NewBalanceChecker(apiKey, evmAddress, solanaAddress string) *BalanceChecker // GetAllBalances fetches balances for all configured wallets // Returns map[chain][token] = balance_usd func (bc *BalanceChecker) GetAllBalances() (map[string]map[string]float64, error) { + balances, _, err := bc.GetAllBalancesDetailed() + return balances, err +} + +// GetAllBalancesDetailed fetches balances for all configured wallets. +// Primary source is the Mobula portfolio API; on per-chain failure it falls +// back to direct RPC reads of the triangle tokens. The degraded flag is true +// when at least one chain could not be read by EITHER path: callers must treat +// degraded balances as unreadable (never "empty wallet") and must not trigger +// automatic rebalancing from them. +func (bc *BalanceChecker) GetAllBalancesDetailed() (map[string]map[string]float64, bool, error) { result := make(map[string]map[string]float64) // Initialize chains @@ -60,11 +86,17 @@ func (bc *BalanceChecker) GetAllBalances() (map[string]map[string]float64, error result["Base"] = make(map[string]float64) result["Arbitrum"] = make(map[string]float64) + degraded := false + // Fetch Solana balances if bc.solanaAddress != "" { solBalances, err := bc.fetchWalletBalance(bc.solanaAddress) if err != nil { log.Printf("⚠️ Failed to fetch Solana balances: %v", err) + if !bc.fillSolanaFromChain(result) { + log.Printf("⚠️ On-chain Solana fallback also failed: balances degraded") + degraded = true + } } else { indexAssets(result, solBalances, "Solana") } @@ -76,6 +108,10 @@ func (bc *BalanceChecker) GetAllBalances() (map[string]map[string]float64, error baseBalances, err := bc.fetchWalletBalanceByChain(bc.evmAddress, "base") if err != nil { log.Printf("⚠️ Failed to fetch Base balances: %v", err) + if !bc.fillEVMFromChain(result, "Base") { + log.Printf("⚠️ On-chain Base fallback also failed: balances degraded") + degraded = true + } } else { indexAssets(result, baseBalances, "Base") } @@ -84,12 +120,118 @@ func (bc *BalanceChecker) GetAllBalances() (map[string]map[string]float64, error arbBalances, err := bc.fetchWalletBalanceByChain(bc.evmAddress, "arbitrum") if err != nil { log.Printf("⚠️ Failed to fetch Arbitrum balances: %v", err) + if !bc.fillEVMFromChain(result, "Arbitrum") { + log.Printf("⚠️ On-chain Arbitrum fallback also failed: balances degraded") + degraded = true + } } else { indexAssets(result, arbBalances, "Arbitrum") } } - return result, nil + if degraded { + bridgeBalanceReadDegraded.Set(1) + } else { + bridgeBalanceReadDegraded.Set(0) + } + + return result, degraded, nil +} + +// Token addresses read by the on-chain fallback. Stablecoins are valued at $1 +// parity (good enough to gate executions); SOL and ETH go through the price +// cache with a static fallback because during a Mobula outage the pricer may +// be down too, and a stale gas estimate beats a zero. +const ( + solanaUSDCMint = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" + baseUSDCAddr = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" + arbUSDTAddr = "0xFd086bC7CD5C481DCC9C85ebE478A1C0b69FCbb9" + arbUSDCAddr = "0xaf88d065e77c8cc2239327c5edb3a432268e5831" +) + +// fillSolanaFromChain reads Solana USDC + native SOL via RPC. Returns true only +// when the triangle token (USDC) was read successfully: SOL alone is not enough +// to run the cycle simulation, so a USDC read failure keeps the chain degraded. +func (bc *BalanceChecker) fillSolanaFromChain(result map[string]map[string]float64) bool { + if bc.onchain == nil || bc.onchain.solanaClient == nil { + return false + } + owner, err := solana.PublicKeyFromBase58(bc.solanaAddress) + if err != nil { + return false + } + + mint := solana.MustPublicKeyFromBase58(solanaUSDCMint) + raw, err := bc.onchain.solanaSPLBalanceOf(owner, mint) + if err != nil { + return false + } + usd := rawToFloat(raw, 6) + result["Solana"]["USDC"] = usd + result["Solana"][strings.ToLower(solanaUSDCMint)] = usd + + if lamports, err := bc.onchain.solanaNativeBalance(owner); err == nil { + result["Solana"]["SOL"] = rawToFloat(lamports, 9) * TokenPriceUSD("SOL", 150) + } + log.Printf("✅ Solana balances recovered via on-chain fallback (USDC $%.2f)", usd) + return true +} + +// fillEVMFromChain reads the chain's triangle stablecoin + native ETH via RPC. +// Same success rule as Solana: the triangle token read must succeed. +func (bc *BalanceChecker) fillEVMFromChain(result map[string]map[string]float64, chain string) bool { + if bc.onchain == nil { + return false + } + owner := common.HexToAddress(bc.evmAddress) + + type tokenRead struct { + addr string + symbols []string + } + var reads []tokenRead + switch chain { + case "Base": + reads = []tokenRead{{baseUSDCAddr, []string{"USDC"}}} + case "Arbitrum": + // Both USDT0 and USDT symbols: cycle_sim reads "USDT0" (Mobula naming) + // while route helpers fall back to "USDT". + reads = []tokenRead{ + {arbUSDTAddr, []string{"USDT0", "USDT"}}, + {arbUSDCAddr, []string{"USDC"}}, + } + default: + return false + } + + ok := false + for i, r := range reads { + raw, err := bc.onchain.erc20BalanceOf(chain, common.HexToAddress(r.addr), owner) + if err != nil { + // Only the first entry is the triangle token; secondary reads are + // best-effort for the stranded-fund gauges. + if i == 0 { + return false + } + continue + } + usd := rawToFloat(raw, 6) + for _, sym := range r.symbols { + result[chain][sym] = usd + } + result[chain][strings.ToLower(r.addr)] = usd + if i == 0 { + ok = true + } + } + + if wei, err := bc.onchain.evmNativeBalance(chain, owner); err == nil { + result[chain]["ETH"] = rawToFloat(wei, 18) * TokenPriceUSD("ETH", 3000) + } + if ok { + log.Printf("✅ %s balances recovered via on-chain fallback", chain) + } + return ok } // indexAssets stores balances in the result map, keyed by BOTH symbol AND contract address (lowercase). diff --git a/harnesses/bridge-monitor/cmd/monitor/executor.go b/harnesses/bridge-monitor/cmd/monitor/executor.go index 94ef212b..34dfde66 100644 --- a/harnesses/bridge-monitor/cmd/monitor/executor.go +++ b/harnesses/bridge-monitor/cmd/monitor/executor.go @@ -112,6 +112,11 @@ func NewExecutor( } else { e.txExecutor = txExec log.Println("✅ TxExecutor initialized") + // Give the balance checker an RPC fallback so a Mobula portfolio + // API outage no longer reads as an empty wallet. + if balanceCheck != nil { + balanceCheck.SetOnchainFallback(txExec) + } } } diff --git a/harnesses/bridge-monitor/cmd/monitor/metrics.go b/harnesses/bridge-monitor/cmd/monitor/metrics.go index 0ac368ae..7d60c219 100644 --- a/harnesses/bridge-monitor/cmd/monitor/metrics.go +++ b/harnesses/bridge-monitor/cmd/monitor/metrics.go @@ -128,4 +128,12 @@ var ( Name: "bridge_consecutive_failures", Help: "Number of consecutive execution failures for a bridge (resets on success)", }, []string{"bridge", "region"}) + + // 1 when at least one chain's balances could not be read by either the + // Mobula API or the on-chain fallback. Lets alerting distinguish "wallet is + // empty" from "we cannot see the wallet". + bridgeBalanceReadDegraded = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bridge_balance_read_degraded", + Help: "1 if wallet balances are currently unreadable via both API and RPC, 0 otherwise", + }) ) diff --git a/harnesses/bridge-monitor/cmd/monitor/onchain_balance.go b/harnesses/bridge-monitor/cmd/monitor/onchain_balance.go index ac450ef1..c6aac96d 100644 --- a/harnesses/bridge-monitor/cmd/monitor/onchain_balance.go +++ b/harnesses/bridge-monitor/cmd/monitor/onchain_balance.go @@ -12,6 +12,7 @@ import ( "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/ethclient" "github.com/gagliardetto/solana-go" "github.com/gagliardetto/solana-go/rpc" ) @@ -49,6 +50,27 @@ func (tx *TxExecutor) evmClientFor(chain string) interface{ CallContract(context return nil } +// evmNativeBalance returns the native (ETH) balance of owner in wei. Needed by +// the balance fallback path: the Mobula portfolio API is the primary source, but +// when it is down we still need gas balances to gate executions safely. +func (tx *TxExecutor) evmNativeBalance(chain string, owner common.Address) (*big.Int, error) { + var client *ethclient.Client + switch strings.ToLower(chain) { + case "base": + client = tx.baseClient + case "arbitrum": + client = tx.arbitrumClient + default: + return nil, fmt.Errorf("unknown EVM chain: %s", chain) + } + if client == nil { + return nil, fmt.Errorf("no RPC client for %s", chain) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + return client.BalanceAt(ctx, owner, nil) +} + // solanaSPLBalanceOf returns the SPL token balance for `owner` and `mint`. If // the associated token account does not yet exist (never received this token), // returns 0 — that's a valid pre-execution state. From 90658e072f92493e7decbb203c24be323ead88a2 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:07:07 +0200 Subject: [PATCH 02/13] bridge-monitor: self-healing metrics + slack notifiers --- .../bridge-monitor/cmd/monitor/metrics.go | 50 +++++++++++++++ harnesses/bridge-monitor/cmd/monitor/slack.go | 62 +++++++++++++++++++ 2 files changed, 112 insertions(+) diff --git a/harnesses/bridge-monitor/cmd/monitor/metrics.go b/harnesses/bridge-monitor/cmd/monitor/metrics.go index 7d60c219..5f8ec776 100644 --- a/harnesses/bridge-monitor/cmd/monitor/metrics.go +++ b/harnesses/bridge-monitor/cmd/monitor/metrics.go @@ -136,4 +136,54 @@ var ( Name: "bridge_balance_read_degraded", Help: "1 if wallet balances are currently unreadable via both API and RPC, 0 otherwise", }) + + // Auto-rebalance attempts by outcome: attempted, succeeded, failed, capped. + bridgeRebalanceAttempts = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bridge_rebalance_attempts_total", + Help: "Total automatic rebalance attempts by outcome (attempted, succeeded, failed, capped)", + }, []string{"outcome"}) + + // Set to 1 when a scheduled tier was downgraded to a smaller amount because + // the requested tier was not viable even after rebalancing. Reset to 0 when + // the original tier runs at full size again. + bridgeTierDowngraded = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "bridge_tier_downgraded", + Help: "1 when the last run of tier `from` was downgraded to tier `to`, 0 otherwise", + }, []string{"from", "to"}) + + // Hours funds have been sitting off their home triangle leg. 0 when the + // wallet only holds expected inventory. Exported continuously (also while + // paused) so Prometheus alerting can fire without any execution enabled. + bridgeStrandedHours = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "bridge_stranded_hours", + Help: "Hours a balance has been stranded off its home triangle leg (0 when home)", + }, []string{"chain", "token"}) + + // Gas top-up attempts per chain by outcome: attempted, succeeded, failed, + // capped, gated (needed but GAS_TOPUP_ENABLED is off). + bridgeGasTopup = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bridge_gas_topup_total", + Help: "Total gas top-up attempts by chain and outcome", + }, []string{"chain", "outcome"}) ) + +// initSelfHealingMetrics pre-seeds the label combinations the alerting rules +// query, so the series exist on /metrics from process start instead of only +// after the first event. +func initSelfHealingMetrics() { + for _, outcome := range []string{"attempted", "succeeded", "failed", "capped"} { + bridgeRebalanceAttempts.WithLabelValues(outcome).Add(0) + } + bridgeTierDowngraded.WithLabelValues("300", "50").Set(0) + bridgeTierDowngraded.WithLabelValues("300", "5").Set(0) + bridgeTierDowngraded.WithLabelValues("50", "5").Set(0) + bridgeStrandedHours.WithLabelValues("Solana", "USDC").Set(0) + bridgeStrandedHours.WithLabelValues("Base", "USDC").Set(0) + bridgeStrandedHours.WithLabelValues("Arbitrum", "USDT0").Set(0) + for _, chain := range []string{"Solana", "Base", "Arbitrum"} { + for _, outcome := range []string{"attempted", "succeeded", "failed", "capped", "gated"} { + bridgeGasTopup.WithLabelValues(chain, outcome).Add(0) + } + } + bridgeBalanceReadDegraded.Set(0) +} diff --git a/harnesses/bridge-monitor/cmd/monitor/slack.go b/harnesses/bridge-monitor/cmd/monitor/slack.go index 7b9422f1..5b42f773 100644 --- a/harnesses/bridge-monitor/cmd/monitor/slack.go +++ b/harnesses/bridge-monitor/cmd/monitor/slack.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "net/http" + "strings" "time" ) @@ -278,6 +279,67 @@ Waiting for next scheduled slot. Top up the blocked leg to unblock. return s.send(message) } +// NotifyRebalance reports every auto-rebalance event (attempted, succeeded, +// failed, capped) so the owner can audit what the self-healing loop moved. +func (s *SlackNotifier) NotifyRebalance(outcome, detail string) error { + if s == nil { + return nil + } + emoji := map[string]string{ + "attempted": "🔧", + "succeeded": "✅", + "failed": "❌", + "capped": "🛑", + }[outcome] + if emoji == "" { + emoji = "🔧" + } + message := fmt.Sprintf(`%s *Auto-Rebalance %s* + +%s`, emoji, strings.ToUpper(outcome), detail) + return s.send(message) +} + +// NotifyTierDowngrade reports that a scheduled tier ran at a smaller amount +// because the requested size was not viable. Partial data beats none, but the +// owner should know the wallet needs attention. +func (s *SlackNotifier) NotifyTierDowngrade(from, to float64, reason string) error { + if s == nil { + return nil + } + message := fmt.Sprintf(`⬇️ *Tier DOWNGRADED: $%.0f to $%.0f* + +*Reason $%.0f was blocked:* %s + +Running the smaller tier so the benchmark keeps producing data.`, + from, to, from, reason) + return s.send(message) +} + +// NotifyReaper reports stuck-fund reaper actions (corrective transfer of funds +// stranded off their home triangle leg). +func (s *SlackNotifier) NotifyReaper(detail string) error { + if s == nil { + return nil + } + message := fmt.Sprintf(`🧹 *Stuck-Fund Reaper* + +%s`, detail) + return s.send(message) +} + +// NotifyGasTopUp reports gas top-up events (low native gas detected, swap +// attempted, succeeded, failed, capped or gated). +func (s *SlackNotifier) NotifyGasTopUp(chain, outcome, detail string) error { + if s == nil { + return nil + } + message := fmt.Sprintf(`⛽ *Gas Top-Up %s on %s* + +%s`, strings.ToUpper(outcome), chain, detail) + return s.send(message) +} + // NotifyStartup sends a startup notification func (s *SlackNotifier) NotifyStartup(mode string) error { if s == nil { From 2283c9781703355b41a1e35c543b42106fc44ea1 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:08:48 +0200 Subject: [PATCH 03/13] bridge-monitor: auto-rebalancer, corrective transfer via cheapest bridge --- .../bridge-monitor/cmd/monitor/rebalancer.go | 308 ++++++++++++++++++ 1 file changed, 308 insertions(+) create mode 100644 harnesses/bridge-monitor/cmd/monitor/rebalancer.go diff --git a/harnesses/bridge-monitor/cmd/monitor/rebalancer.go b/harnesses/bridge-monitor/cmd/monitor/rebalancer.go new file mode 100644 index 00000000..27d815ba --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/rebalancer.go @@ -0,0 +1,308 @@ +package main + +import ( + "fmt" + "log" + "strings" + "time" +) + +// Auto-rebalancer: when the pre-flight cycle simulation says a tier cannot run +// because one leg is short, move the shortfall from the leg holding excess +// inventory instead of waiting for a human. Reuses the executor's quote and +// broadcast plumbing so a rebalance transfer is measured and notified exactly +// like a benchmark execution. cmd/rebalance stays as the manual fallback. + +const ( + // Two attempts per tier slot: one transfer plus one retry covers transient + // quote failures without letting a broken bridge drain the daily budget. + maxRebalanceAttemptsPerSlot = 2 + + // Move 10 percent more than the computed shortfall so bridge fees and + // slippage on the transfer itself do not leave the leg short again. + rebalanceBufferFactor = 1.10 +) + +// legSpec describes one home leg of the USDC triangle. +type legSpec struct { + Chain string + ChainAPI string + Token string + TokenAddr string +} + +// The triangle's expected inventory distribution. Order matches R1, R2, R3. +var triangleLegs = []legSpec{ + {Chain: "Solana", ChainAPI: "solana:solana", Token: "USDC", TokenAddr: solanaUSDCMint}, + {Chain: "Base", ChainAPI: "evm:8453", Token: "USDC", TokenAddr: baseUSDCAddr}, + {Chain: "Arbitrum", ChainAPI: "evm:42161", Token: "USDT", TokenAddr: arbUSDTAddr}, +} + +func findLeg(chain string) (legSpec, bool) { + for _, leg := range triangleLegs { + if strings.EqualFold(leg.Chain, chain) { + return leg, true + } + } + return legSpec{}, false +} + +// legBalanceUSD looks up a leg's balance by contract address first (robust +// against symbol drift like USDT vs USDT0 in Mobula's DB), then by symbol. +func legBalanceUSD(balances map[string]map[string]float64, leg legSpec) float64 { + chain := balances[leg.Chain] + if chain == nil { + return 0 + } + if v, ok := chain[strings.ToLower(leg.TokenAddr)]; ok { + return v + } + if v, ok := chain[leg.Token]; ok { + return v + } + if leg.Token == "USDT" { + if v, ok := chain["USDT0"]; ok { + return v + } + } + return 0 +} + +// BuildRefillRoute turns a failed CycleSimulation into a corrective TestRoute. +// The source leg is the triangle leg holding the largest surplus above its own +// 1x tier requirement: taking from anywhere else would just move the blockage. +// The amount is the shortfall plus the fee buffer, never more, so a rebalance +// cannot silently drain a healthy leg. Pure function, unit-tested. +func BuildRefillRoute(sim CycleSimulation, balances map[string]map[string]float64) (TestRoute, float64, error) { + if sim.Viable { + return TestRoute{}, 0, fmt.Errorf("cycle already viable, nothing to rebalance") + } + if sim.RefillUSD <= 0 { + return TestRoute{}, 0, fmt.Errorf("simulation has no refill amount") + } + dest, ok := findLeg(sim.RefillChain) + if !ok { + return TestRoute{}, 0, fmt.Errorf("unknown refill chain %q", sim.RefillChain) + } + + var source legSpec + bestSurplus := 0.0 + for _, leg := range triangleLegs { + if leg.Chain == dest.Chain { + continue + } + surplus := legBalanceUSD(balances, leg) - sim.Tier + if surplus > bestSurplus { + source = leg + bestSurplus = surplus + } + } + if bestSurplus <= 0 { + return TestRoute{}, 0, fmt.Errorf("no leg holds excess inventory above its own $%.0f tier need", sim.Tier) + } + if bestSurplus < sim.RefillUSD { + return TestRoute{}, 0, fmt.Errorf("no leg holds enough excess: need $%.2f, best surplus is $%.2f on %s", + sim.RefillUSD, bestSurplus, source.Chain) + } + + amount := sim.RefillUSD * rebalanceBufferFactor + if amount > bestSurplus { + amount = bestSurplus + } + + route := TestRoute{ + Name: fmt.Sprintf("REBALANCE_%s_%s", strings.ToUpper(source.Chain), strings.ToUpper(dest.Chain)), + FromChain: source.Chain, + FromChainAPI: source.ChainAPI, + FromToken: source.TokenAddr, + ToChain: dest.Chain, + ToChainAPI: dest.ChainAPI, + ToToken: dest.TokenAddr, + IsSolanaSrc: source.Chain == "Solana", + } + return route, amount, nil +} + +// Rebalancer executes corrective transfers via the cheapest live bridge quote. +type Rebalancer struct { + executor *Executor + slack *SlackNotifier +} + +func NewRebalancer(executor *Executor, slack *SlackNotifier) *Rebalancer { + if executor == nil { + return nil + } + return &Rebalancer{executor: executor, slack: slack} +} + +// canAct verifies keys exist and we are in a mode allowed to broadcast. Keeps +// the rebalancer inert in dry-run and while the benchmark is paused. +func (r *Rebalancer) canAct() bool { + return r != nil && r.executor != nil && r.executor.txExecutor != nil && r.executor.txExecutor.CanExecute() +} + +// TryUnblockTier attempts up to maxRebalanceAttemptsPerSlot corrective +// transfers to make the given failed simulation viable. Returns the latest +// simulation and whether the tier can now run. Callers must only pass +// non-degraded balances: rebalancing on unreadable balances could move real +// funds based on phantom zeros. +func (r *Rebalancer) TryUnblockTier(sim CycleSimulation, balances map[string]map[string]float64, tierLabel string) (CycleSimulation, bool) { + if !r.canAct() { + return sim, false + } + + for attempt := 1; attempt <= maxRebalanceAttemptsPerSlot; attempt++ { + if r.executor.config.DailySpentUSD >= r.executor.config.MaxDailySpendUSD { + bridgeRebalanceAttempts.WithLabelValues("capped").Inc() + _ = r.slack.NotifyRebalance("capped", fmt.Sprintf( + "Tier $%.0f (%s) blocked (%s) but daily spend limit is reached ($%.2f / $%.2f). No rebalance attempted.", + sim.Tier, tierLabel, sim.Reason, r.executor.config.DailySpentUSD, r.executor.config.MaxDailySpendUSD)) + return sim, false + } + + route, amountUSD, err := BuildRefillRoute(sim, balances) + if err != nil { + bridgeRebalanceAttempts.WithLabelValues("failed").Inc() + log.Printf("🔧 Rebalance not possible for tier $%.0f: %v", sim.Tier, err) + _ = r.slack.NotifyRebalance("failed", fmt.Sprintf( + "Tier $%.0f (%s) blocked (%s) and no corrective route could be built: %v", + sim.Tier, tierLabel, sim.Reason, err)) + return sim, false + } + + bridgeRebalanceAttempts.WithLabelValues("attempted").Inc() + log.Printf("🔧 Rebalance attempt %d/%d: $%.2f %s -> %s (unblocks tier $%.0f)", + attempt, maxRebalanceAttemptsPerSlot, amountUSD, route.FromChain, route.ToChain, sim.Tier) + _ = r.slack.NotifyRebalance("attempted", fmt.Sprintf( + "Tier $%.0f (%s) blocked: %s\nMoving $%.2f from %s %s to %s %s (attempt %d/%d).", + sim.Tier, tierLabel, sim.Reason, amountUSD, route.FromChain, sourceSymbol(route), route.ToChain, sim.RefillToken, + attempt, maxRebalanceAttemptsPerSlot)) + + result := r.ExecuteCheapest(route, amountUSD) + if result == nil || !result.Success { + bridgeRebalanceAttempts.WithLabelValues("failed").Inc() + detail := "no bridge produced a usable quote" + if result != nil && result.Error != nil { + detail = result.Error.Error() + } + _ = r.slack.NotifyRebalance("failed", fmt.Sprintf( + "Rebalance transfer for tier $%.0f failed: %s", sim.Tier, detail)) + } else { + bridgeRebalanceAttempts.WithLabelValues("succeeded").Inc() + r.executor.config.DailySpentUSD += result.ActualFeeUSD + _ = r.slack.NotifyRebalance("succeeded", fmt.Sprintf( + "Moved $%.2f from %s to %s via %s (fee $%.4f, tx %s). Re-checking tier $%.0f viability.", + amountUSD, route.FromChain, route.ToChain, result.Bridge, result.ActualFeeUSD, result.TxHash, sim.Tier)) + } + + // Give the destination credit a moment to be indexed by the portfolio + // API before re-simulating, otherwise we would re-read the old state. + time.Sleep(15 * time.Second) + + fresh, degraded, err := r.executor.balanceCheck.GetAllBalancesDetailed() + if err != nil || degraded { + log.Printf("🔧 Post-rebalance balances unreadable (degraded=%v err=%v), stopping", degraded, err) + return sim, false + } + balances = fresh + sim = SimulateTriangleCycle(balances, sim.Tier) + if sim.Viable { + return sim, true + } + } + + bridgeRebalanceAttempts.WithLabelValues("capped").Inc() + _ = r.slack.NotifyRebalance("capped", fmt.Sprintf( + "Tier $%.0f still not viable after %d rebalance attempts: %s", + sim.Tier, maxRebalanceAttemptsPerSlot, sim.Reason)) + return sim, false +} + +// ExecuteCheapest quotes all three executing bridges and broadcasts through +// the cheapest one that returned a live quote. Shared by the tier rebalancer +// and the stuck-fund reaper so both follow the same cost discipline. +func (r *Rebalancer) ExecuteCheapest(route TestRoute, amountUSD float64) *ExecutionResult { + bridge, fee, err := r.quoteCheapest(route, amountUSD) + if err != nil { + log.Printf("🔧 Cheapest-quote selection failed for %s: %v", route.Name, err) + return nil + } + log.Printf("🔧 Cheapest bridge for %s: %s (quoted fee $%.4f)", route.Name, bridge, fee) + + rawUnits := toRawUnits(amountUSD) + return r.executor.executeOnBridge(bridge, route, amountUSD, amountUSD, rawUnits) +} + +// quoteCheapest fetches quotes from Mobula, Relay and LI.FI and returns the +// bridge with the lowest total quoted fee among those that answered. +func (r *Rebalancer) quoteCheapest(route TestRoute, amountUSD float64) (string, float64, error) { + e := r.executor + sender := e.walletManager.EVMAddress + receiver := e.walletManager.EVMAddress + if route.FromChain == "Solana" { + sender = e.walletManager.SolanaAddress + } + if route.ToChain == "Solana" { + receiver = e.walletManager.SolanaAddress + } + rawUnits := toRawUnits(amountUSD) + + bestBridge := "" + bestFee := 0.0 + + consider := func(bridge string, fee float64, err error) { + if err != nil { + log.Printf("🔧 [%s] rebalance quote failed: %v", bridge, err) + return + } + if bestBridge == "" || fee < bestFee { + bestBridge, bestFee = bridge, fee + } + } + + if e.mobula != nil { + quote, _, err := e.mobula.GetQuote(route.FromChainAPI, route.FromToken, route.ToChainAPI, route.ToToken, sender, receiver, amountUSD) + if err != nil { + consider("mobula", 0, err) + } else { + consider("mobula", parseFloatOrZero(quote.Data.Fees.TotalFeeUsd)+parseFloatOrZero(quote.Data.Fees.GasFeeUsd), nil) + } + } + + if quote, _, err := e.relay.GetQuote(route, rawUnits, sender, receiver); err != nil { + consider("relay", 0, err) + } else { + consider("relay", parseFloatOrZero(quote.Fees.RelayerService.AmountUsd)+parseFloatOrZero(quote.Fees.RelayerGas.AmountUsd), nil) + } + + if quote, _, err := e.lifi.GetQuote(route, rawUnits, sender, receiver); err != nil { + consider("lifi", 0, err) + } else { + fee := 0.0 + for _, f := range quote.Estimate.FeeCosts { + fee += parseFloatOrZero(f.AmountUSD) + } + consider("lifi", fee, nil) + } + + if bestBridge == "" { + return "", 0, fmt.Errorf("all bridges failed to quote %s", route.Name) + } + return bestBridge, bestFee, nil +} + +func parseFloatOrZero(s string) float64 { + return parseFloat(s, 0) +} + +// sourceSymbol maps the route's source token address back to a human symbol +// for Slack messages. +func sourceSymbol(route TestRoute) string { + for _, leg := range triangleLegs { + if strings.EqualFold(leg.TokenAddr, route.FromToken) { + return leg.Token + } + } + return route.FromToken +} From ca1efe8a0f60f4f1131ccea5d7669f98f72d81e0 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:10:21 +0200 Subject: [PATCH 04/13] bridge-monitor: tier downgrade ladder + rebalancer wiring in pre-flight --- harnesses/bridge-monitor/cmd/monitor/main.go | 117 +++++++++++++++---- 1 file changed, 94 insertions(+), 23 deletions(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index 5f867994..ee504f81 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -139,6 +139,9 @@ func main() { } } + // Pre-seed the self-healing series so alerting rules match from startup. + initSelfHealingMetrics() + // Start Prometheus metrics endpoint. Railway / most PaaS inject $PORT // and route external traffic to whatever value they chose. If we bind // to the wrong port the edge proxy returns 502 with x-railway-fallback. @@ -336,6 +339,14 @@ func main() { log.Println("🚀 Production mode: fixed-time scheduler started") } + // Auto-rebalancer shares the executor's quote and broadcast plumbing. Inert + // unless the executor can actually broadcast (production mode, keys present), + // so it costs nothing in dry-run or while paused. + var rebalancer *Rebalancer + if executor != nil { + rebalancer = NewRebalancer(executor, slackNotifier) + } + // Track last meme execution day to run weekly lastMemeDay := -1 @@ -345,7 +356,7 @@ func main() { case <-getSchedulerChan(scheduler, "$5"): // $5 execution loop - daily at 10:00 UTC if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, GetTriangleRoutes(), 5.0, "daily") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 5.0, "daily") // Meme routes use independent capital (TRUMP) — always attempt, // the per-route RunReal check catches insufficient TRUMP. @@ -361,49 +372,110 @@ func main() { case <-getSchedulerChan(scheduler, "$50"): if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, GetTriangleRoutes(), 50.0, "Mon+Thu") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 50.0, "Mon+Thu") } case <-getSchedulerChan(scheduler, "$300"): if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, GetTriangleRoutes(), 300.0, "Mon weekly") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 300.0, "Mon weekly") } } } } +// downgradeLadder returns the tier amounts to try, largest first, starting at +// the scheduled tier. Partial data beats none: if $300 cannot run we still +// want the $50 or $5 datapoint from the same slot. +func downgradeLadder(tier float64) []float64 { + all := []float64{300, 50, 5} + var out []float64 + for _, t := range all { + if t <= tier { + out = append(out, t) + } + } + if len(out) == 0 { + out = []float64{tier} + } + return out +} + // runTierIfViable pre-flights the full R1→R2→R3 cycle at the given tier. If the -// simulation says the cycle cannot complete, emit ONE Slack "couldn't run" message -// and skip — next scheduler tick will retry. Returns true if the tier actually ran. +// simulation says the cycle cannot complete, it first lets the auto-rebalancer +// try to unblock the tier, then walks the downgrade ladder to a smaller amount. +// Only if nothing on the ladder is viable does it emit ONE Slack "couldn't run" +// message and skip — next scheduler tick will retry. Returns true if any +// amount actually ran. func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifier, - routes []TestRoute, tier float64, tierLabel string, + rebalancer *Rebalancer, routes []TestRoute, tier float64, tierLabel string, ) bool { if bc == nil { log.Printf("⚠️ No balance checker — skipping tier $%.0f pre-flight", tier) return false } - balances, err := bc.GetAllBalances() - if err != nil { - log.Printf("⚠️ Pre-flight balance fetch failed for $%.0f tier: %v", tier, err) - if slack != nil { - _ = slack.NotifyTierSkipped(tier, tierLabel, fmt.Sprintf("Balance API error: %v", err)) + ladder := downgradeLadder(tier) + blockedReason := "" + + for i, amount := range ladder { + balances, degraded, err := bc.GetAllBalancesDetailed() + if err != nil || degraded { + // Unreadable is not the same as empty: refuse to act (and above all + // refuse to rebalance) on balances we cannot trust. + log.Printf("⚠️ Pre-flight balances unreadable for $%.0f tier (degraded=%v err=%v)", tier, degraded, err) + if slack != nil { + _ = slack.NotifyTierSkipped(tier, tierLabel, + fmt.Sprintf("Balances unreadable (degraded=%v, err=%v). Refusing to run or rebalance blind.", degraded, err)) + } + return false } - return false - } - sim := SimulateTriangleCycle(balances, tier) - if !sim.Viable { - log.Printf("⏭️ Tier $%.0f skipped: %s", tier, sim.Reason) - if slack != nil { - _ = slack.NotifyTierSkipped(tier, tierLabel, sim.Reason) + sim := SimulateTriangleCycle(balances, amount) + if !sim.Viable && rebalancer != nil { + sim, _ = rebalancer.TryUnblockTier(sim, balances, tierLabel) } - return false + if !sim.Viable { + if blockedReason == "" { + blockedReason = sim.Reason + } + log.Printf("⏭️ Tier $%.0f not viable at $%.0f: %s", tier, amount, sim.Reason) + continue + } + + if i > 0 { + log.Printf("⬇️ Tier $%.0f downgraded to $%.0f: %s", tier, amount, blockedReason) + bridgeTierDowngraded.WithLabelValues(tierAmountLabel(tier), tierAmountLabel(amount)).Set(1) + if slack != nil { + _ = slack.NotifyTierDowngrade(tier, amount, blockedReason) + } + } else { + // The scheduled tier runs at full size again: clear its downgrade + // gauges so the alert stops firing. + for _, smaller := range ladder[1:] { + bridgeTierDowngraded.WithLabelValues(tierAmountLabel(tier), tierAmountLabel(smaller)).Set(0) + } + } + + runTriangle(executor, routes, amount) + return true + } + + log.Printf("⏭️ Tier $%.0f skipped entirely: %s", tier, blockedReason) + if slack != nil { + _ = slack.NotifyTierSkipped(tier, tierLabel, blockedReason) } + return false +} + +func tierAmountLabel(tier float64) string { + return strconv.FormatFloat(tier, 'f', 0, 64) +} - // Sequential per-bridge orchestration: each bridge runs a full R1→R2→R3 triangle - // before the next bridge starts. Halves the peak capital need per leg, gives a - // clean round-trip cost per provider, and unlocks larger tiers with less capital. +// runTriangle runs the sequential per-bridge orchestration: each bridge runs a +// full R1→R2→R3 triangle before the next bridge starts. Halves the peak capital +// need per leg, gives a clean round-trip cost per provider, and unlocks larger +// tiers with less capital. +func runTriangle(executor *Executor, routes []TestRoute, tier float64) { log.Printf("💸 Running $%.0f triangle (sequential per-bridge)...", tier) bridges := []string{"mobula", "relay", "lifi"} for _, bridge := range bridges { @@ -423,7 +495,6 @@ func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifie } } } - return true } // getSchedulerChan returns the appropriate scheduler channel or nil From 91656a967addeacc3b5e2393ce80b3aa9bc15282 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:11:34 +0200 Subject: [PATCH 05/13] bridge-monitor: stuck-fund reaper, hourly stranded gauge + corrective transfer --- harnesses/bridge-monitor/cmd/monitor/main.go | 4 + .../bridge-monitor/cmd/monitor/reaper.go | 251 ++++++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 harnesses/bridge-monitor/cmd/monitor/reaper.go diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index ee504f81..18c5815a 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -347,6 +347,10 @@ func main() { rebalancer = NewRebalancer(executor, slackNotifier) } + // Stuck-fund reaper: gauges in every mode (alerting must survive a pause), + // corrective transfers only in production and unpaused. + StartReaper(balanceChecker, rebalancer, slackNotifier, config.ExecutionMode, paused) + // Track last meme execution day to run weekly lastMemeDay := -1 diff --git a/harnesses/bridge-monitor/cmd/monitor/reaper.go b/harnesses/bridge-monitor/cmd/monitor/reaper.go new file mode 100644 index 00000000..da6cec88 --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/reaper.go @@ -0,0 +1,251 @@ +package main + +import ( + "fmt" + "log" + "os" + "strings" + "time" +) + +// Stuck-fund reaper: the triangle expects inventory on exactly three legs +// (Sol USDC, Base USDC, Arb USDT) plus native gas. Refunds and partial fills +// sometimes leave capital elsewhere (typically USDC on Arbitrum after an R3 +// refund), where no scheduled route will ever pick it up again. The reaper +// tracks how long such balances sit stranded, exports that as a gauge for +// alerting, and, when execution is enabled, sends ONE corrective transfer per +// hourly tick back to the neediest home leg. + +// Ignore balances below this: bridge dust and rounding leftovers are not worth +// a corrective transfer's fees. +const strandedDustUSD = 1.0 + +// Reaper transfers share the rebalancer's spirit of "never move more than a +// tier plus buffer": one capped hop per tick instead of one big blind sweep. +const reaperMaxTransferUSD = 300 * rebalanceBufferFactor + +// Tokens that are SUPPOSED to sit on each chain. Everything else above the +// dust floor counts as stranded. TRUMP and BRETT are meme-route inventory, +// native SOL and ETH are gas. +var homeTokens = map[string]map[string]bool{ + "Solana": {"USDC": true, "SOL": true, "TRUMP": true}, + "Base": {"USDC": true, "ETH": true, "BRETT": true}, + "Arbitrum": {"USDT0": true, "USDT": true, "ETH": true}, +} + +// Stranded tokens we know how to route home. Anything else is alert-only: +// building a bridge TX for an unknown asset from inside a repair loop is how +// funds get burned, so we only ever move assets we have addresses for. +var movableStranded = map[string]map[string]string{ + "Solana": {"USDT": "Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB"}, + "Base": {"USDT": "0xfde4C96c8593536E31F229EA8f37b2ADa2699bb2"}, + "Arbitrum": {"USDC": arbUSDCAddr}, +} + +type strandKey struct { + Chain string + Token string +} + +// StrandedTracker remembers when each off-home balance was first seen so the +// stranded duration survives across ticks (but not restarts: a restart resets +// the clock, which only delays the reaper, never makes it over-eager). +type StrandedTracker struct { + firstSeen map[strandKey]time.Time +} + +func NewStrandedTracker() *StrandedTracker { + return &StrandedTracker{firstSeen: make(map[strandKey]time.Time)} +} + +// Update computes stranded hours for every off-home balance in the snapshot +// and returns them, including explicit zeros for keys that just came home so +// the exported gauge resets. Pure given (balances, now), unit-tested. +func (t *StrandedTracker) Update(balances map[string]map[string]float64, now time.Time) map[strandKey]float64 { + hours := make(map[strandKey]float64) + + // Zero out anything previously tracked; re-add below if still stranded. + for k := range t.firstSeen { + hours[k] = 0 + } + + for chain, tokens := range balances { + home := homeTokens[chain] + for token, usd := range tokens { + // Contract-address keys duplicate the symbol entries. + if strings.HasPrefix(token, "0x") || len(token) > 12 { + continue + } + if usd <= strandedDustUSD || (home != nil && home[token]) { + delete(t.firstSeen, strandKey{chain, token}) + continue + } + k := strandKey{chain, token} + first, seen := t.firstSeen[k] + if !seen { + first = now + t.firstSeen[k] = first + } + hours[k] = now.Sub(first).Hours() + } + } + + // Drop firstSeen entries that vanished from the snapshot entirely. + for k := range t.firstSeen { + if _, still := hours[k]; !still { + delete(t.firstSeen, k) + } + } + for k, h := range hours { + if h == 0 { + delete(t.firstSeen, k) + } + } + return hours +} + +// StartReaper launches the hourly stuck-fund loop. The gauge side runs in +// every mode, including while BENCHMARK_PAUSED, because alerting must keep +// working during a pause. The corrective transfer only fires in production +// mode, unpaused, with broadcast-capable keys. +func StartReaper(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifier, mode string, paused bool) { + if bc == nil { + log.Println("🧹 Reaper disabled: no balance checker") + return + } + + strandedHours := parseFloat(os.Getenv("REAPER_STRANDED_HOURS"), 6) + tracker := NewStrandedTracker() + + canAct := mode == "production" && !paused && rebalancer.canAct() + log.Printf("🧹 Stuck-fund reaper started (threshold %.0fh, acting=%v)", strandedHours, canAct) + + go func() { + // First evaluation immediately so gauges exist right after boot, then hourly. + reaperTick(bc, rebalancer, slack, tracker, strandedHours, canAct) + ticker := time.NewTicker(1 * time.Hour) + defer ticker.Stop() + for range ticker.C { + reaperTick(bc, rebalancer, slack, tracker, strandedHours, canAct) + } + }() +} + +func reaperTick(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifier, + tracker *StrandedTracker, thresholdHours float64, canAct bool, +) { + balances, degraded, err := bc.GetAllBalancesDetailed() + if err != nil || degraded { + // Without trustworthy balances we cannot tell stranded from home, so + // neither the gauge nor a transfer would mean anything this tick. + log.Printf("🧹 Reaper tick skipped: balances unreadable (degraded=%v err=%v)", degraded, err) + return + } + + hours := tracker.Update(balances, time.Now().UTC()) + + // Home legs always report 0 so alert expressions have a baseline series. + for _, leg := range triangleLegs { + sym := leg.Token + if leg.Chain == "Arbitrum" { + sym = "USDT0" + } + bridgeStrandedHours.WithLabelValues(leg.Chain, sym).Set(0) + } + for k, h := range hours { + bridgeStrandedHours.WithLabelValues(k.Chain, k.Token).Set(h) + } + + // Pick the single worst offender above threshold; one transfer per tick + // keeps the blast radius of a bad tick to one capped TX. + var worst strandKey + worstHours := 0.0 + for k, h := range hours { + if h >= thresholdHours && h > worstHours { + worst, worstHours = k, h + } + } + if worstHours == 0 { + log.Printf("🧹 Reaper tick: no funds stranded beyond %.0fh", thresholdHours) + return + } + + amount := balances[worst.Chain][worst.Token] + log.Printf("🧹 Stranded funds: $%.2f %s on %s for %.1fh (threshold %.0fh)", + amount, worst.Token, worst.Chain, worstHours, thresholdHours) + + if !canAct { + // Read-only mode: the gauge above is the alert path, a human handles it. + return + } + + srcAddr, known := movableStranded[worst.Chain][worst.Token] + if !known { + _ = slack.NotifyReaper(fmt.Sprintf( + "$%.2f of %s stranded on %s for %.1fh, but I have no safe route for that asset. Manual rebalance needed (cmd/rebalance).", + amount, worst.Token, worst.Chain, worstHours)) + return + } + + if rebalancer.executor.config.DailySpentUSD >= rebalancer.executor.config.MaxDailySpendUSD { + _ = slack.NotifyReaper(fmt.Sprintf( + "$%.2f of %s stranded on %s for %.1fh, but daily spend limit is reached ($%.2f / $%.2f). Will retry next tick.", + amount, worst.Token, worst.Chain, worstHours, + rebalancer.executor.config.DailySpentUSD, rebalancer.executor.config.MaxDailySpendUSD)) + return + } + + dest := neediestHomeLeg(balances) + if amount > reaperMaxTransferUSD { + amount = reaperMaxTransferUSD + } + + route := TestRoute{ + Name: fmt.Sprintf("REAPER_%s_%s", strings.ToUpper(worst.Chain), strings.ToUpper(dest.Chain)), + FromChain: worst.Chain, + FromChainAPI: chainAPIFor(worst.Chain), + FromToken: srcAddr, + ToChain: dest.Chain, + ToChainAPI: dest.ChainAPI, + ToToken: dest.TokenAddr, + IsSolanaSrc: worst.Chain == "Solana", + } + + _ = slack.NotifyReaper(fmt.Sprintf( + "Moving $%.2f of stranded %s from %s (stuck %.1fh) home to %s %s via cheapest bridge.", + amount, worst.Token, worst.Chain, worstHours, dest.Chain, dest.Token)) + + result := rebalancer.ExecuteCheapest(route, amount) + if result == nil || !result.Success { + detail := "no bridge produced a usable quote" + if result != nil && result.Error != nil { + detail = result.Error.Error() + } + _ = slack.NotifyReaper(fmt.Sprintf("Corrective transfer FAILED: %s. Will retry next tick.", detail)) + return + } + + rebalancer.executor.config.DailySpentUSD += result.ActualFeeUSD + _ = slack.NotifyReaper(fmt.Sprintf( + "Corrective transfer succeeded via %s (fee $%.4f, tx %s).", result.Bridge, result.ActualFeeUSD, result.TxHash)) +} + +// neediestHomeLeg returns the triangle leg with the lowest balance: stranded +// funds should land where they unblock the next scheduled cycle soonest. +func neediestHomeLeg(balances map[string]map[string]float64) legSpec { + best := triangleLegs[0] + bestBal := legBalanceUSD(balances, best) + for _, leg := range triangleLegs[1:] { + if b := legBalanceUSD(balances, leg); b < bestBal { + best, bestBal = leg, b + } + } + return best +} + +func chainAPIFor(chain string) string { + if leg, ok := findLeg(chain); ok { + return leg.ChainAPI + } + return "" +} From bc7e169de911170204be45ce07f5d9531a2ec11b Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:13:19 +0200 Subject: [PATCH 06/13] bridge-monitor: gas auto-top-up via LiFi same-chain swap, gated by default --- .../bridge-monitor/cmd/monitor/gas_topup.go | 197 ++++++++++++++++++ harnesses/bridge-monitor/cmd/monitor/main.go | 16 +- 2 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 harnesses/bridge-monitor/cmd/monitor/gas_topup.go diff --git a/harnesses/bridge-monitor/cmd/monitor/gas_topup.go b/harnesses/bridge-monitor/cmd/monitor/gas_topup.go new file mode 100644 index 00000000..1a9d2148 --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/gas_topup.go @@ -0,0 +1,197 @@ +package main + +import ( + "fmt" + "log" + "os" + "strings" + "time" +) + +// Gas auto-top-up: an execution slot can fail for the dumbest possible reason, +// no native gas to sign the deposit TX. Pre-flight already fetches balances, +// so we check SOL / Base ETH / Arb ETH against a USD floor and, when allowed, +// swap wallet USDC into native gas on that same chain via a LI.FI same-chain +// swap (their /v1/quote accepts fromChain == toChain and returns broadcastable +// calldata, same shape as a bridge quote). +// +// SAFETY GATE: GAS_TOPUP_ENABLED defaults to false. The plumbing below reuses +// the proven LiFi execution path, but the same-chain variant has not been +// exercised with real funds from this harness yet. Until someone runs one +// supervised top-up per chain and flips the env var, low gas only alerts. + +// LiFi native-token markers. +const ( + lifiEVMNative = "0x0000000000000000000000000000000000000000" + lifiSolanaNative = "11111111111111111111111111111111" +) + +type gasChain struct { + Chain string + NativeSym string + USDCSource string +} + +var gasChains = []gasChain{ + {Chain: "Solana", NativeSym: "SOL", USDCSource: solanaUSDCMint}, + {Chain: "Base", NativeSym: "ETH", USDCSource: baseUSDCAddr}, + {Chain: "Arbitrum", NativeSym: "ETH", USDCSource: arbUSDCAddr}, +} + +type GasTopper struct { + executor *Executor + slack *SlackNotifier + + enabled bool + minUSD float64 + topupUSD float64 + + // Per-chain USD swapped today; resets on UTC day change. Caps a runaway + // price-feed glitch to one top-up per chain per day. + spentToday map[string]float64 + day int +} + +func NewGasTopper(executor *Executor, slack *SlackNotifier) *GasTopper { + if executor == nil { + return nil + } + return &GasTopper{ + executor: executor, + slack: slack, + enabled: strings.EqualFold(strings.TrimSpace(os.Getenv("GAS_TOPUP_ENABLED")), "true"), + minUSD: parseFloat(os.Getenv("GAS_MIN_USD"), 15), + topupUSD: parseFloat(os.Getenv("GAS_TOPUP_USD"), 25), + spentToday: make(map[string]float64), + day: time.Now().UTC().YearDay(), + } +} + +// CheckAndTopUp inspects native gas on each chain and tops up where needed. +// Called from tier pre-flight with balances the caller already validated as +// non-degraded: a phantom zero here would otherwise trigger a pointless swap. +func (g *GasTopper) CheckAndTopUp(balances map[string]map[string]float64) { + if g == nil { + return + } + + if today := time.Now().UTC().YearDay(); today != g.day { + g.day = today + g.spentToday = make(map[string]float64) + } + + for _, gc := range gasChains { + gasUSD := balances[gc.Chain][gc.NativeSym] + if gasUSD >= g.minUSD { + continue + } + log.Printf("⛽ Low gas on %s: $%.2f %s (floor $%.2f)", gc.Chain, gasUSD, gc.NativeSym, g.minUSD) + + if !g.enabled { + bridgeGasTopup.WithLabelValues(gc.Chain, "gated").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "gated", fmt.Sprintf( + "Native gas is $%.2f, below the $%.2f floor, but GAS_TOPUP_ENABLED is off. Top up manually or enable after a supervised test.", + gasUSD, g.minUSD)) + continue + } + if g.executor.txExecutor == nil || !g.executor.txExecutor.CanExecute() { + // Dry-run or missing keys: detection is still useful in logs, but + // there is nothing safe to broadcast. + continue + } + if g.spentToday[gc.Chain] >= g.topupUSD { + bridgeGasTopup.WithLabelValues(gc.Chain, "capped").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "capped", fmt.Sprintf( + "Native gas is $%.2f but today's top-up budget ($%.2f) is already spent.", gasUSD, g.topupUSD)) + continue + } + + amount := g.topupUSD - g.spentToday[gc.Chain] + if amount > g.topupUSD { + amount = g.topupUSD + } + g.topUpChain(gc, amount, gasUSD) + } +} + +// topUpChain swaps `amountUSD` of USDC into native gas on one chain. +func (g *GasTopper) topUpChain(gc gasChain, amountUSD, gasUSD float64) { + bridgeGasTopup.WithLabelValues(gc.Chain, "attempted").Inc() + + toToken := lifiEVMNative + sender := g.executor.walletManager.EVMAddress + if gc.Chain == "Solana" { + toToken = lifiSolanaNative + sender = g.executor.walletManager.SolanaAddress + } + + route := TestRoute{ + Name: fmt.Sprintf("GAS_TOPUP_%s", strings.ToUpper(gc.Chain)), + FromChain: gc.Chain, + FromChainAPI: chainAPIFor(gc.Chain), + FromToken: gc.USDCSource, + ToChain: gc.Chain, + ToChainAPI: chainAPIFor(gc.Chain), + ToToken: toToken, + IsSolanaSrc: gc.Chain == "Solana", + } + rawUnits := toRawUnits(amountUSD) + + quote, _, err := g.executor.lifi.GetQuote(route, rawUnits, sender, sender) + if err != nil { + bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", fmt.Sprintf( + "LI.FI same-chain swap quote failed (gas $%.2f, wanted $%.2f USDC to %s): %v", gasUSD, amountUSD, gc.NativeSym, err)) + return + } + + // EVM swaps of ERC-20 input need the router approved first, same as bridges. + if quote.Estimate.ApprovalAddress != "" && gc.Chain != "Solana" { + approvalHash, err := g.executor.txExecutor.ApproveERC20(gc.Chain, quote.Action.FromToken.Address, quote.Estimate.ApprovalAddress, quote.Action.FromAmount) + if err != nil { + bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", fmt.Sprintf("USDC approval failed: %v", err)) + return + } + time.Sleep(5 * time.Second) + if ok, err := g.executor.txExecutor.CheckEVMTxStatus(gc.Chain, approvalHash); err != nil || !ok { + bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", "USDC approval not confirmed") + return + } + } + + var txHash string + if gc.Chain == "Solana" { + txHash, err = g.executor.txExecutor.ExecuteSolanaTransaction(quote.TransactionRequest.Data) + } else { + txHash, err = g.executor.txExecutor.ExecuteEVMTransaction(gc.Chain, quote.TransactionRequest.To, quote.TransactionRequest.Data, quote.TransactionRequest.Value) + } + if err != nil { + bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", fmt.Sprintf("Broadcast failed: %v", err)) + return + } + + // Same-chain swaps settle in one TX: a confirmed receipt is a fill, no + // bridge status polling needed. Solana broadcast acceptance is our signal. + if gc.Chain != "Solana" { + time.Sleep(8 * time.Second) + if ok, err := g.executor.txExecutor.CheckEVMTxStatus(gc.Chain, txHash); err != nil || !ok { + bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", fmt.Sprintf("Swap TX not confirmed or reverted: %s", txHash)) + return + } + } + + g.spentToday[gc.Chain] += amountUSD + fee := 0.0 + for _, f := range quote.Estimate.FeeCosts { + fee += parseFloatOrZero(f.AmountUSD) + } + g.executor.config.DailySpentUSD += fee + bridgeGasTopup.WithLabelValues(gc.Chain, "succeeded").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "succeeded", fmt.Sprintf( + "Swapped $%.2f USDC to %s (was $%.2f, fee $%.4f, tx %s). Daily budget used: $%.2f / $%.2f.", + amountUSD, gc.NativeSym, gasUSD, fee, txHash, g.spentToday[gc.Chain], g.topupUSD)) +} diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index 18c5815a..d7d7eaa4 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -343,8 +343,10 @@ func main() { // unless the executor can actually broadcast (production mode, keys present), // so it costs nothing in dry-run or while paused. var rebalancer *Rebalancer + var gasTopper *GasTopper if executor != nil { rebalancer = NewRebalancer(executor, slackNotifier) + gasTopper = NewGasTopper(executor, slackNotifier) } // Stuck-fund reaper: gauges in every mode (alerting must survive a pause), @@ -360,7 +362,7 @@ func main() { case <-getSchedulerChan(scheduler, "$5"): // $5 execution loop - daily at 10:00 UTC if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 5.0, "daily") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, gasTopper, GetTriangleRoutes(), 5.0, "daily") // Meme routes use independent capital (TRUMP) — always attempt, // the per-route RunReal check catches insufficient TRUMP. @@ -376,12 +378,12 @@ func main() { case <-getSchedulerChan(scheduler, "$50"): if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 50.0, "Mon+Thu") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, gasTopper, GetTriangleRoutes(), 50.0, "Mon+Thu") } case <-getSchedulerChan(scheduler, "$300"): if executor != nil && config.ExecutionMode == "production" { - runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, GetTriangleRoutes(), 300.0, "Mon weekly") + runTierIfViable(executor, balanceChecker, slackNotifier, rebalancer, gasTopper, GetTriangleRoutes(), 300.0, "Mon weekly") } } } @@ -411,7 +413,7 @@ func downgradeLadder(tier float64) []float64 { // message and skip — next scheduler tick will retry. Returns true if any // amount actually ran. func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifier, - rebalancer *Rebalancer, routes []TestRoute, tier float64, tierLabel string, + rebalancer *Rebalancer, gasTopper *GasTopper, routes []TestRoute, tier float64, tierLabel string, ) bool { if bc == nil { log.Printf("⚠️ No balance checker — skipping tier $%.0f pre-flight", tier) @@ -434,6 +436,12 @@ func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifie return false } + // Gas check once per slot: the same balances snapshot already carries + // SOL / ETH, and an execution without gas fails later anyway. + if i == 0 { + gasTopper.CheckAndTopUp(balances) + } + sim := SimulateTriangleCycle(balances, amount) if !sim.Viable && rebalancer != nil { sim, _ = rebalancer.TryUnblockTier(sim, balances, tierLabel) From 9626ac4abfff89f4615a44c607996df21e7ed039 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:14:35 +0200 Subject: [PATCH 07/13] bridge-monitor: unit tests for refill route, ladder, stranded hours; fix tracker reset --- .../bridge-monitor/cmd/monitor/reaper.go | 20 +- .../cmd/monitor/selfheal_test.go | 182 ++++++++++++++++++ 2 files changed, 189 insertions(+), 13 deletions(-) create mode 100644 harnesses/bridge-monitor/cmd/monitor/selfheal_test.go diff --git a/harnesses/bridge-monitor/cmd/monitor/reaper.go b/harnesses/bridge-monitor/cmd/monitor/reaper.go index da6cec88..c2ea8eee 100644 --- a/harnesses/bridge-monitor/cmd/monitor/reaper.go +++ b/harnesses/bridge-monitor/cmd/monitor/reaper.go @@ -63,11 +63,7 @@ func NewStrandedTracker() *StrandedTracker { // the exported gauge resets. Pure given (balances, now), unit-tested. func (t *StrandedTracker) Update(balances map[string]map[string]float64, now time.Time) map[strandKey]float64 { hours := make(map[strandKey]float64) - - // Zero out anything previously tracked; re-add below if still stranded. - for k := range t.firstSeen { - hours[k] = 0 - } + current := make(map[strandKey]bool) for chain, tokens := range balances { home := homeTokens[chain] @@ -77,10 +73,10 @@ func (t *StrandedTracker) Update(balances map[string]map[string]float64, now tim continue } if usd <= strandedDustUSD || (home != nil && home[token]) { - delete(t.firstSeen, strandKey{chain, token}) continue } k := strandKey{chain, token} + current[k] = true first, seen := t.firstSeen[k] if !seen { first = now @@ -90,14 +86,12 @@ func (t *StrandedTracker) Update(balances map[string]map[string]float64, now tim } } - // Drop firstSeen entries that vanished from the snapshot entirely. + // Keys that just came home report an explicit 0 (so the exported gauge + // resets) and lose their firstSeen entry (so a later re-strand restarts + // the clock instead of inheriting the old one). for k := range t.firstSeen { - if _, still := hours[k]; !still { - delete(t.firstSeen, k) - } - } - for k, h := range hours { - if h == 0 { + if !current[k] { + hours[k] = 0 delete(t.firstSeen, k) } } diff --git a/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go new file mode 100644 index 00000000..1731bb75 --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go @@ -0,0 +1,182 @@ +package main + +import ( + "math" + "testing" + "time" +) + +func balancesSnapshot(sol, base, arb float64) map[string]map[string]float64 { + return map[string]map[string]float64{ + "Solana": {"USDC": sol, "SOL": 30}, + "Base": {"USDC": base, "ETH": 40}, + "Arbitrum": {"USDT0": arb, "ETH": 25}, + } +} + +func TestBuildRefillRouteSourceSelection(t *testing.T) { + // Solana leg is short for a $50 tier; Base holds the biggest surplus so it + // must be picked as the source, not Arbitrum. + balances := balancesSnapshot(10, 400, 120) + sim := SimulateTriangleCycle(balances, 50) + if sim.Viable { + t.Fatal("expected $50 cycle to be blocked on Solana") + } + if sim.RefillChain != "Solana" { + t.Fatalf("expected refill chain Solana, got %s", sim.RefillChain) + } + + route, amount, err := BuildRefillRoute(sim, balances) + if err != nil { + t.Fatalf("BuildRefillRoute failed: %v", err) + } + if route.FromChain != "Base" { + t.Errorf("expected source Base (largest surplus), got %s", route.FromChain) + } + if route.ToChain != "Solana" { + t.Errorf("expected destination Solana, got %s", route.ToChain) + } + if route.FromToken != baseUSDCAddr { + t.Errorf("expected Base USDC source token, got %s", route.FromToken) + } + if route.ToToken != solanaUSDCMint { + t.Errorf("expected Solana USDC destination token, got %s", route.ToToken) + } + if !route.IsSolanaSrc == (route.FromChain == "Solana") { + t.Error("IsSolanaSrc inconsistent with FromChain") + } + + // Shortfall is 50 - 10 = 40, buffered by 10 percent. + want := sim.RefillUSD * rebalanceBufferFactor + if math.Abs(amount-want) > 0.01 { + t.Errorf("expected amount %.2f (need + 10%% buffer), got %.2f", want, amount) + } + if amount > 400-50 { + t.Errorf("amount %.2f exceeds Base surplus", amount) + } +} + +func TestBuildRefillRouteCapsAtSurplus(t *testing.T) { + // Surplus barely covers the shortfall: the buffered amount must be clamped + // to the surplus instead of overdrawing the source leg. + balances := balancesSnapshot(0, 55.5, 5) + sim := SimulateTriangleCycle(balances, 5) + if sim.Viable { + t.Fatal("expected $5 cycle to be blocked on Solana") + } + route, amount, err := BuildRefillRoute(sim, balances) + if err != nil { + t.Fatalf("BuildRefillRoute failed: %v", err) + } + if route.FromChain != "Base" { + t.Fatalf("expected source Base, got %s", route.FromChain) + } + surplus := 55.5 - 5 + if amount > surplus+0.001 { + t.Errorf("amount %.2f exceeds surplus %.2f", amount, surplus) + } +} + +func TestBuildRefillRouteNoExcess(t *testing.T) { + // Every leg is broke: refusing is the only safe answer. + balances := balancesSnapshot(1, 2, 3) + sim := SimulateTriangleCycle(balances, 50) + if _, _, err := BuildRefillRoute(sim, balances); err == nil { + t.Fatal("expected error when no leg holds excess inventory") + } +} + +func TestBuildRefillRouteViableRejected(t *testing.T) { + balances := balancesSnapshot(500, 500, 500) + sim := SimulateTriangleCycle(balances, 50) + if !sim.Viable { + t.Fatal("expected viable cycle") + } + if _, _, err := BuildRefillRoute(sim, balances); err == nil { + t.Fatal("expected error for viable simulation") + } +} + +func TestDowngradeLadder(t *testing.T) { + cases := []struct { + tier float64 + want []float64 + }{ + {300, []float64{300, 50, 5}}, + {50, []float64{50, 5}}, + {5, []float64{5}}, + } + for _, c := range cases { + got := downgradeLadder(c.tier) + if len(got) != len(c.want) { + t.Fatalf("tier %.0f: got %v want %v", c.tier, got, c.want) + } + for i := range got { + if got[i] != c.want[i] { + t.Errorf("tier %.0f: got %v want %v", c.tier, got, c.want) + break + } + } + } +} + +func TestStrandedHoursComputation(t *testing.T) { + tracker := NewStrandedTracker() + t0 := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC) + + // USDC on Arbitrum is off-home (triangle expects USDT there). + balances := balancesSnapshot(100, 100, 100) + balances["Arbitrum"]["USDC"] = 42 + + hours := tracker.Update(balances, t0) + k := strandKey{"Arbitrum", "USDC"} + if hours[k] != 0 { + t.Errorf("first sighting should report 0 hours, got %.2f", hours[k]) + } + // Home legs never appear as stranded. + if _, ok := hours[strandKey{"Solana", "USDC"}]; ok { + t.Error("home leg Solana/USDC reported as stranded") + } + if _, ok := hours[strandKey{"Arbitrum", "USDT0"}]; ok { + t.Error("home leg Arbitrum/USDT0 reported as stranded") + } + + // 7 hours later the same balance is still there: above the 6h default. + hours = tracker.Update(balances, t0.Add(7*time.Hour)) + if math.Abs(hours[k]-7) > 0.001 { + t.Errorf("expected 7 stranded hours, got %.2f", hours[k]) + } + + // Funds moved home: the key must report an explicit 0 so the gauge resets, + // and the clock must restart if it strands again later. + delete(balances["Arbitrum"], "USDC") + hours = tracker.Update(balances, t0.Add(8*time.Hour)) + if hours[k] != 0 { + t.Errorf("expected explicit 0 after funds came home, got %.2f", hours[k]) + } + + balances["Arbitrum"]["USDC"] = 42 + hours = tracker.Update(balances, t0.Add(20*time.Hour)) + if hours[k] != 0 { + t.Errorf("re-stranding must restart the clock at 0, got %.2f", hours[k]) + } +} + +func TestStrandedIgnoresDustAndAddressKeys(t *testing.T) { + tracker := NewStrandedTracker() + now := time.Now().UTC() + + balances := balancesSnapshot(100, 100, 100) + balances["Arbitrum"]["USDC"] = 0.5 + balances["Arbitrum"][arbUSDCAddr] = 5000 + + hours := tracker.Update(balances, now) + if _, ok := hours[strandKey{"Arbitrum", "USDC"}]; ok { + t.Error("dust below $1 must not count as stranded") + } + for k := range hours { + if len(k.Token) > 12 { + t.Errorf("contract-address key leaked into stranded set: %v", k) + } + } +} From 3b55ed372d834bb8986c0149e5d7fbbe9359e964 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:16:06 +0200 Subject: [PATCH 08/13] bridge-monitor: keyless dry-run support for reaper + RunDryRun --- .../bridge-monitor/cmd/monitor/executor.go | 19 +++++++++++++----- .../bridge-monitor/cmd/monitor/reaper.go | 20 ++++++++++++++----- 2 files changed, 29 insertions(+), 10 deletions(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/executor.go b/harnesses/bridge-monitor/cmd/monitor/executor.go index 34dfde66..d88a0770 100644 --- a/harnesses/bridge-monitor/cmd/monitor/executor.go +++ b/harnesses/bridge-monitor/cmd/monitor/executor.go @@ -133,12 +133,21 @@ func (e *Executor) RunDryRun(route TestRoute, amountUSD float64) *ExecutionResul log.Printf("🧪 [DRY-RUN] Testing %s with $%.0f", route.Name, amountUSD) - // Step 1: Check balances + // Step 1: Check balances. Keyless dry-run has no balance checker, so fall + // back to the SIMULATE_BALANCES snapshot instead of crashing. log.Printf(" 📊 Checking balances...") - balances, err := e.balanceCheck.GetAllBalances() - if err != nil { - log.Printf(" ❌ Balance check failed: %v", err) - result.Error = err + var balances map[string]map[string]float64 + if e.balanceCheck != nil { + var err error + balances, err = e.balanceCheck.GetAllBalances() + if err != nil { + log.Printf(" ❌ Balance check failed: %v", err) + result.Error = err + return result + } + } else if balances = SimulateBalances(); balances == nil { + log.Printf(" ❌ No balance checker and SIMULATE_BALANCES not set") + result.Error = fmt.Errorf("no balance source in dry-run") return result } diff --git a/harnesses/bridge-monitor/cmd/monitor/reaper.go b/harnesses/bridge-monitor/cmd/monitor/reaper.go index c2ea8eee..d40f5f6a 100644 --- a/harnesses/bridge-monitor/cmd/monitor/reaper.go +++ b/harnesses/bridge-monitor/cmd/monitor/reaper.go @@ -103,7 +103,17 @@ func (t *StrandedTracker) Update(balances map[string]map[string]float64, now tim // working during a pause. The corrective transfer only fires in production // mode, unpaused, with broadcast-capable keys. func StartReaper(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifier, mode string, paused bool) { - if bc == nil { + // Balance source: real checker in normal operation, the SIMULATE_BALANCES + // snapshot in keyless dry-run so the gauge path stays testable locally. + var fetch func() (map[string]map[string]float64, bool, error) + switch { + case bc != nil: + fetch = bc.GetAllBalancesDetailed + case SimulateBalances() != nil: + fetch = func() (map[string]map[string]float64, bool, error) { + return SimulateBalances(), false, nil + } + default: log.Println("🧹 Reaper disabled: no balance checker") return } @@ -116,19 +126,19 @@ func StartReaper(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifie go func() { // First evaluation immediately so gauges exist right after boot, then hourly. - reaperTick(bc, rebalancer, slack, tracker, strandedHours, canAct) + reaperTick(fetch, rebalancer, slack, tracker, strandedHours, canAct) ticker := time.NewTicker(1 * time.Hour) defer ticker.Stop() for range ticker.C { - reaperTick(bc, rebalancer, slack, tracker, strandedHours, canAct) + reaperTick(fetch, rebalancer, slack, tracker, strandedHours, canAct) } }() } -func reaperTick(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifier, +func reaperTick(fetch func() (map[string]map[string]float64, bool, error), rebalancer *Rebalancer, slack *SlackNotifier, tracker *StrandedTracker, thresholdHours float64, canAct bool, ) { - balances, degraded, err := bc.GetAllBalancesDetailed() + balances, degraded, err := fetch() if err != nil || degraded { // Without trustworthy balances we cannot tell stranded from home, so // neither the gauge nor a transfer would mean anything this tick. From ed084d84d4464a0ec1ff99a11dd17c61531b4a9e Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 15:18:47 +0200 Subject: [PATCH 09/13] bridge-monitor: comment style cleanup in pre-flight docs --- harnesses/bridge-monitor/cmd/monitor/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index d7d7eaa4..037ee994 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -410,7 +410,7 @@ func downgradeLadder(tier float64) []float64 { // simulation says the cycle cannot complete, it first lets the auto-rebalancer // try to unblock the tier, then walks the downgrade ladder to a smaller amount. // Only if nothing on the ladder is viable does it emit ONE Slack "couldn't run" -// message and skip — next scheduler tick will retry. Returns true if any +// message and skip, so the next scheduler tick retries. Returns true if any // amount actually ran. func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifier, rebalancer *Rebalancer, gasTopper *GasTopper, routes []TestRoute, tier float64, tierLabel string, From d2c3d0037a3e2caacd404799456474e790008e1d Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 16:03:26 +0200 Subject: [PATCH 10/13] bridge-monitor: daily spend tracker keyed by UTC date, persisted; execution mutex primitive DailySpentUSD only ever incremented and was zeroed by every restart, so a crash-loop minted a fresh budget each time. The tracker resets exactly at UTC midnight, persists (date, spent) to SPEND_STATE_PATH (default ./spend-state.json, not /tmp) and is mutex-guarded. Also declares the package-level single-flight lock and the FAILED_TX_FEE_ESTIMATE_USD (default 0.50) helper for gas bled by failed broadcasts. --- harnesses/bridge-monitor/.gitignore | 2 + .../cmd/monitor/spend_tracker.go | 124 ++++++++++++++++++ .../cmd/monitor/spend_tracker_test.go | 79 +++++++++++ 3 files changed, 205 insertions(+) create mode 100644 harnesses/bridge-monitor/cmd/monitor/spend_tracker.go create mode 100644 harnesses/bridge-monitor/cmd/monitor/spend_tracker_test.go diff --git a/harnesses/bridge-monitor/.gitignore b/harnesses/bridge-monitor/.gitignore index 091b86f3..ad59826f 100644 --- a/harnesses/bridge-monitor/.gitignore +++ b/harnesses/bridge-monitor/.gitignore @@ -1,5 +1,7 @@ .env bin/ +spend-state.json +spend-state.json.tmp *.log prometheus_data/ grafana_data/ diff --git a/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go b/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go new file mode 100644 index 00000000..7a885299 --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go @@ -0,0 +1,124 @@ +package main + +import ( + "encoding/json" + "log" + "os" + "strings" + "sync" + "time" +) + +// execMu is the package-level single-flight lock for every code path that can +// broadcast a transaction: scheduled triangle runs, the auto-rebalancer, the +// stuck-fund reaper and gas top-ups. All of these actors sign with the same +// two wallets, so two concurrent broadcasts race on PendingNonceAt and can +// reuse a nonce. Scheduler slots take the lock for the whole slot; the reaper +// only TryLocks and skips its corrective action when the lock is busy, it +// never queues behind a running slot. +var execMu sync.Mutex + +// SpendTracker accounts spending against the daily cap. The counter is keyed +// by UTC date so the budget resets exactly at midnight UTC, and the +// (date, spent) pair is persisted to disk so a process restart within the +// same UTC day keeps the consumed budget instead of minting a fresh one +// (a crash-looping process used to reset the counter on every start). +type SpendTracker struct { + mu sync.Mutex + path string + now func() time.Time + date string + spent float64 +} + +type spendState struct { + Date string `json:"date"` + SpentUSD float64 `json:"spent_usd"` +} + +// NewSpendTracker loads persisted state from path. A nil clock uses time.Now; +// tests inject a fake clock to drive the date rollover. State recorded on a +// different UTC date is discarded on load. +func NewSpendTracker(path string, now func() time.Time) *SpendTracker { + if now == nil { + now = time.Now + } + t := &SpendTracker{path: path, now: now} + t.date = t.utcDate() + if data, err := os.ReadFile(path); err == nil { + var s spendState + if json.Unmarshal(data, &s) == nil && s.Date == t.date && s.SpentUSD > 0 { + t.spent = s.SpentUSD + log.Printf("💾 Loaded daily spend state: $%.2f already spent on %s (%s)", t.spent, t.date, path) + } + } + return t +} + +func (t *SpendTracker) utcDate() string { + return t.now().UTC().Format("2006-01-02") +} + +// rolloverLocked resets the counter when the UTC date has changed since the +// last access. Callers must hold t.mu. +func (t *SpendTracker) rolloverLocked() { + if d := t.utcDate(); d != t.date { + log.Printf("💾 Daily spend reset: %s -> %s (was $%.2f)", t.date, d, t.spent) + t.date = d + t.spent = 0 + t.saveLocked() + } +} + +// Add books usd against today's budget and persists the new total. +func (t *SpendTracker) Add(usd float64) { + if usd <= 0 { + return + } + t.mu.Lock() + defer t.mu.Unlock() + t.rolloverLocked() + t.spent += usd + t.saveLocked() +} + +// Spent returns today's consumed budget, applying the UTC rollover first. +func (t *SpendTracker) Spent() float64 { + t.mu.Lock() + defer t.mu.Unlock() + t.rolloverLocked() + return t.spent +} + +// saveLocked persists atomically (write temp, rename). Callers hold t.mu. +func (t *SpendTracker) saveLocked() { + data, err := json.Marshal(spendState{Date: t.date, SpentUSD: t.spent}) + if err != nil { + return + } + tmp := t.path + ".tmp" + if err := os.WriteFile(tmp, data, 0o644); err != nil { + log.Printf("⚠️ Failed to persist spend state to %s: %v", tmp, err) + return + } + if err := os.Rename(tmp, t.path); err != nil { + log.Printf("⚠️ Failed to persist spend state to %s: %v", t.path, err) + } +} + +// spendStatePath resolves where the (date, spent) pair lives. Defaults to a +// path relative to the working directory (NOT /tmp, which some distros wipe +// on a timer) so restarts on the same host see the same file. +func spendStatePath() string { + if p := strings.TrimSpace(os.Getenv("SPEND_STATE_PATH")); p != "" { + return p + } + return "./spend-state.json" +} + +// failedTxFeeEstimateUSD is booked against the daily cap for any broadcast +// that produced a TxHash but never confirmed as a success: the deposit or +// approval TX most likely paid gas even though the bridge never filled. +func failedTxFeeEstimateUSD() float64 { + return parseFloat(os.Getenv("FAILED_TX_FEE_ESTIMATE_USD"), 0.50) +} diff --git a/harnesses/bridge-monitor/cmd/monitor/spend_tracker_test.go b/harnesses/bridge-monitor/cmd/monitor/spend_tracker_test.go new file mode 100644 index 00000000..f7637d8e --- /dev/null +++ b/harnesses/bridge-monitor/cmd/monitor/spend_tracker_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "math" + "os" + "path/filepath" + "testing" + "time" +) + +func TestSpendTrackerUTCRollover(t *testing.T) { + path := filepath.Join(t.TempDir(), "spend-state.json") + now := time.Date(2026, 7, 12, 23, 0, 0, 0, time.UTC) + clock := func() time.Time { return now } + + tr := NewSpendTracker(path, clock) + tr.Add(3.5) + if got := tr.Spent(); math.Abs(got-3.5) > 1e-9 { + t.Fatalf("expected 3.5 spent, got %v", got) + } + + // A restart within the same UTC day must keep the consumed budget: a + // crash-loop must not mint a fresh budget. + tr2 := NewSpendTracker(path, clock) + if got := tr2.Spent(); math.Abs(got-3.5) > 1e-9 { + t.Fatalf("restart lost same-day spend: got %v, want 3.5", got) + } + + // Crossing midnight UTC resets the counter without a restart. + now = time.Date(2026, 7, 13, 1, 0, 0, 0, time.UTC) + if got := tr2.Spent(); got != 0 { + t.Fatalf("expected 0 after UTC date rollover, got %v", got) + } + tr2.Add(1.25) + + // A restart on the new day loads the new day's spend. + tr3 := NewSpendTracker(path, clock) + if got := tr3.Spent(); math.Abs(got-1.25) > 1e-9 { + t.Fatalf("restart on new day: got %v, want 1.25", got) + } +} + +func TestSpendTrackerDiscardsStaleState(t *testing.T) { + path := filepath.Join(t.TempDir(), "spend-state.json") + if err := os.WriteFile(path, []byte(`{"date":"2026-07-11","spent_usd":9.99}`), 0o644); err != nil { + t.Fatal(err) + } + clock := func() time.Time { return time.Date(2026, 7, 12, 8, 0, 0, 0, time.UTC) } + tr := NewSpendTracker(path, clock) + if got := tr.Spent(); got != 0 { + t.Fatalf("yesterday's spend must not carry over, got %v", got) + } +} + +func TestSpendTrackerIgnoresCorruptState(t *testing.T) { + path := filepath.Join(t.TempDir(), "spend-state.json") + if err := os.WriteFile(path, []byte("not json"), 0o644); err != nil { + t.Fatal(err) + } + tr := NewSpendTracker(path, nil) + if got := tr.Spent(); got != 0 { + t.Fatalf("corrupt state must read as 0, got %v", got) + } + tr.Add(2) + if got := tr.Spent(); math.Abs(got-2) > 1e-9 { + t.Fatalf("expected 2 after Add, got %v", got) + } +} + +func TestFailedTxFeeEstimateDefault(t *testing.T) { + t.Setenv("FAILED_TX_FEE_ESTIMATE_USD", "") + if got := failedTxFeeEstimateUSD(); math.Abs(got-0.50) > 1e-9 { + t.Fatalf("default estimate must be $0.50, got %v", got) + } + t.Setenv("FAILED_TX_FEE_ESTIMATE_USD", "1.25") + if got := failedTxFeeEstimateUSD(); math.Abs(got-1.25) > 1e-9 { + t.Fatalf("env override not honored, got %v", got) + } +} From b7b3b08e16e0fee07b521928c582da5b025cee84 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 16:04:46 +0200 Subject: [PATCH 11/13] bridge-monitor: fix double-send on in-flight rebalance, slot-wide budget, single-flight lock Fund-loss fixes from hostile review: - a corrective transfer whose deposit broadcast but whose bridge status never resolved is now TERMINAL for the scheduler slot (funds are likely still in flight, a retry double-sends); only pre-broadcast failures may consume another attempt. executeOnBridge keeps TxHash on error paths so callers can tell the two apart. - ONE corrective-transfer budget (max 2) per scheduler slot, shared across all downgrade-ladder rungs (was per rung: up to 6 transfers per slot). - package-level execution mutex: scheduler slots own the wallets end to end, the reaper TryLocks and skips its tick instead of queueing. - broadcasts that failed or timed out after getting a TxHash book a flat FAILED_TX_FEE_ESTIMATE_USD against the daily cap instead of bleeding unaccounted gas; all spend goes through the persisted UTC-day tracker. - gas top-up now honors MaxDailySpendUSD before broadcasting. --- .../bridge-monitor/cmd/monitor/executor.go | 81 +++++++--- .../bridge-monitor/cmd/monitor/gas_topup.go | 17 +- harnesses/bridge-monitor/cmd/monitor/main.go | 17 +- .../bridge-monitor/cmd/monitor/metrics.go | 7 +- .../bridge-monitor/cmd/monitor/reaper.go | 33 ++-- .../bridge-monitor/cmd/monitor/rebalancer.go | 147 ++++++++++++++---- .../cmd/monitor/selfheal_test.go | 69 ++++++++ 7 files changed, 306 insertions(+), 65 deletions(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/executor.go b/harnesses/bridge-monitor/cmd/monitor/executor.go index d88a0770..ca0bd514 100644 --- a/harnesses/bridge-monitor/cmd/monitor/executor.go +++ b/harnesses/bridge-monitor/cmd/monitor/executor.go @@ -18,16 +18,17 @@ const ( ModeProduction ExecutionMode = "production" // Full execution loop ) -// ExecutionConfig holds the execution loop configuration +// ExecutionConfig holds the execution loop configuration. +// Daily spend tracking lives in the Executor's SpendTracker (UTC-date keyed, +// persisted to disk), not here: a plain struct field never reset and was +// zeroed by every restart. type ExecutionConfig struct { - Mode ExecutionMode - Freq5USD time.Duration // How often to run $5 tests - Freq50USD time.Duration // How often to run $50 tests - Freq300USD time.Duration // How often to run $300 tests - EnableDebridge bool // Whether to execute Debridge (expensive) - MaxDailySpendUSD float64 // Safety cap on daily spending - DailySpentUSD float64 // Track daily spending - LastResetDay int // Day of month for daily reset + Mode ExecutionMode + Freq5USD time.Duration // How often to run $5 tests + Freq50USD time.Duration // How often to run $50 tests + Freq300USD time.Duration // How often to run $300 tests + EnableDebridge bool // Whether to execute Debridge (expensive) + MaxDailySpendUSD float64 // Safety cap on daily spending } // ExecutionResult holds the result of an execution test @@ -68,6 +69,36 @@ type Executor struct { debridge *DebridgeBridge region string slack *SlackNotifier + spend *SpendTracker +} + +// DailySpent returns today's consumed budget via the mutex-guarded tracker. +// All cap checks must go through here, never through a raw field, so the +// reaper goroutine and the scheduler loop cannot race on the counter. +func (e *Executor) DailySpent() float64 { + return e.spend.Spent() +} + +// AddDailySpend books usd against today's budget (UTC-date keyed, persisted). +func (e *Executor) AddDailySpend(usd float64) { + e.spend.Add(usd) +} + +// accountSpend books a broadcast's cost against the daily cap. A successful +// fill books the realized fee. Any broadcast that produced a TxHash but did +// not confirm as a success books a conservative flat estimate, because the +// deposit or approval TX most likely burned gas even without a fill. Results +// that never broadcast cost nothing. +func (e *Executor) accountSpend(result *ExecutionResult) { + if result == nil || result.DryRun { + return + } + switch { + case result.Success: + e.AddDailySpend(result.ActualFeeUSD) + case result.TxHash != "": + e.AddDailySpend(failedTxFeeEstimateUSD()) + } } // NewExecutor creates a new executor @@ -92,6 +123,7 @@ func NewExecutor( debridge: debridge, region: region, slack: slack, + spend: NewSpendTracker(spendStatePath(), time.Now), } // Initialize TxExecutor if we have private keys @@ -212,8 +244,8 @@ func (e *Executor) RunReal(route TestRoute, amountUSD float64) []*ExecutionResul } // Check daily spending limit - if e.config.DailySpentUSD >= e.config.MaxDailySpendUSD { - msg := fmt.Sprintf("Daily spending limit reached ($%.2f / $%.2f)", e.config.DailySpentUSD, e.config.MaxDailySpendUSD) + if spent := e.DailySpent(); spent >= e.config.MaxDailySpendUSD { + msg := fmt.Sprintf("Daily spending limit reached ($%.2f / $%.2f)", spent, e.config.MaxDailySpendUSD) log.Printf("⚠️ %s", msg) if e.slack != nil { _ = e.slack.NotifyScheduledSkip(route.Name, route.FromChain, route.FromToken, amountUSD, msg) @@ -275,10 +307,9 @@ func (e *Executor) RunReal(route TestRoute, amountUSD float64) []*ExecutionResul } } - // Update daily spending - if result.Success { - e.config.DailySpentUSD += result.ActualFeeUSD - } + // Update daily spending (realized fee on success, flat gas + // estimate on a broadcast that never confirmed). + e.accountSpend(result) } // Wait between bridges to avoid rate limiting @@ -296,8 +327,8 @@ func (e *Executor) RunBridgeOnRoute(bridge string, route TestRoute, amountUSD fl if e.txExecutor == nil || !e.txExecutor.CanExecute() { return nil } - if e.config.DailySpentUSD >= e.config.MaxDailySpendUSD { - log.Printf("⚠️ Daily spending limit reached ($%.2f / $%.2f)", e.config.DailySpentUSD, e.config.MaxDailySpendUSD) + if spent := e.DailySpent(); spent >= e.config.MaxDailySpendUSD { + log.Printf("⚠️ Daily spending limit reached ($%.2f / $%.2f)", spent, e.config.MaxDailySpendUSD) return nil } @@ -320,9 +351,7 @@ func (e *Executor) RunBridgeOnRoute(bridge string, route TestRoute, amountUSD fl } } - if result.Success { - e.config.DailySpentUSD += result.ActualFeeUSD - } + e.accountSpend(result) return result } @@ -370,13 +399,17 @@ func (e *Executor) executeOnBridge(bridge string, route TestRoute, amount, amoun result.ToToken = route.ToToken result.AmountUSD = amountUSD + // Keep the TxHash even when the attempt errored out: callers use it to + // tell a pre-broadcast failure (safe to retry) from a broadcast whose + // final status is unknown (terminal, funds may still be in flight) and + // to account the gas a failed TX still burned. + result.TxHash = txHash + if err != nil { log.Printf(" ❌ Execution failed: %v", err) result.Error = err return result } - - result.TxHash = txHash // If the sub-function flagged a refund/revert, keep Success=false so Slack and // Prometheus correctly classify it (Reverted takes precedence over Success). result.Success = !result.Reverted @@ -1018,8 +1051,8 @@ func (e *Executor) testBridgeDryRun(bridge string, route TestRoute, amountUSD fl log.Printf(" 💰 Estimated cost: $%.4f", expectedCost) // Update daily spending tracker (even in dry-run for estimation) - e.config.DailySpentUSD += expectedCost - log.Printf(" 📈 Daily spend estimate: $%.2f / $%.2f max", e.config.DailySpentUSD, e.config.MaxDailySpendUSD) + e.AddDailySpend(expectedCost) + log.Printf(" 📈 Daily spend estimate: $%.2f / $%.2f max", e.DailySpent(), e.config.MaxDailySpendUSD) } // ValidateSetup checks that everything is configured correctly diff --git a/harnesses/bridge-monitor/cmd/monitor/gas_topup.go b/harnesses/bridge-monitor/cmd/monitor/gas_topup.go index 1a9d2148..5e244ff6 100644 --- a/harnesses/bridge-monitor/cmd/monitor/gas_topup.go +++ b/harnesses/bridge-monitor/cmd/monitor/gas_topup.go @@ -105,6 +105,15 @@ func (g *GasTopper) CheckAndTopUp(balances map[string]map[string]float64) { "Native gas is $%.2f but today's top-up budget ($%.2f) is already spent.", gasUSD, g.topupUSD)) continue } + // Same daily-cap pre-check the executor applies before any broadcast: + // a top-up is still spend and must not blow past MaxDailySpendUSD. + if spent := g.executor.DailySpent(); spent >= g.executor.config.MaxDailySpendUSD { + bridgeGasTopup.WithLabelValues(gc.Chain, "capped").Inc() + _ = g.slack.NotifyGasTopUp(gc.Chain, "capped", fmt.Sprintf( + "Native gas is $%.2f but the daily spend limit is reached ($%.2f / $%.2f). No top-up attempted.", + gasUSD, spent, g.executor.config.MaxDailySpendUSD)) + continue + } amount := g.topupUSD - g.spentToday[gc.Chain] if amount > g.topupUSD { @@ -156,6 +165,9 @@ func (g *GasTopper) topUpChain(gc gasChain, amountUSD, gasUSD float64) { time.Sleep(5 * time.Second) if ok, err := g.executor.txExecutor.CheckEVMTxStatus(gc.Chain, approvalHash); err != nil || !ok { bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + // The approval TX was broadcast and likely paid gas even though + // it never confirmed: book the conservative estimate. + g.executor.AddDailySpend(failedTxFeeEstimateUSD()) _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", "USDC approval not confirmed") return } @@ -179,6 +191,9 @@ func (g *GasTopper) topUpChain(gc gasChain, amountUSD, gasUSD float64) { time.Sleep(8 * time.Second) if ok, err := g.executor.txExecutor.CheckEVMTxStatus(gc.Chain, txHash); err != nil || !ok { bridgeGasTopup.WithLabelValues(gc.Chain, "failed").Inc() + // Broadcast happened: even a reverted or unconfirmed swap TX + // bled gas, so it counts toward the daily cap. + g.executor.AddDailySpend(failedTxFeeEstimateUSD()) _ = g.slack.NotifyGasTopUp(gc.Chain, "failed", fmt.Sprintf("Swap TX not confirmed or reverted: %s", txHash)) return } @@ -189,7 +204,7 @@ func (g *GasTopper) topUpChain(gc gasChain, amountUSD, gasUSD float64) { for _, f := range quote.Estimate.FeeCosts { fee += parseFloatOrZero(f.AmountUSD) } - g.executor.config.DailySpentUSD += fee + g.executor.AddDailySpend(fee) bridgeGasTopup.WithLabelValues(gc.Chain, "succeeded").Inc() _ = g.slack.NotifyGasTopUp(gc.Chain, "succeeded", fmt.Sprintf( "Swapped $%.2f USDC to %s (was $%.2f, fee $%.4f, tx %s). Daily budget used: $%.2f / $%.2f.", diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index 037ee994..9ed5b3b8 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -247,6 +247,7 @@ func main() { log.Println("⚠️ This will execute REAL transactions!") bridges := []string{"mobula", "relay", "lifi"} + execMu.Lock() for _, bridge := range bridges { log.Printf("\n━━━ Bridge: %s ━━━", bridge) for _, route := range triangleRoutes { @@ -255,6 +256,7 @@ func main() { time.Sleep(2 * time.Second) } } + execMu.Unlock() log.Println("\n✅ Single-test complete! Exiting.") return @@ -369,9 +371,11 @@ func main() { now := time.Now().UTC() if now.Weekday() == time.Monday && now.YearDay() != lastMemeDay { log.Println("💸 Running $5 meme execution tests (weekly)...") + execMu.Lock() for _, route := range GetMemeRoutes() { executor.RunReal(route, 5.0) } + execMu.Unlock() lastMemeDay = now.YearDay() } } @@ -420,9 +424,20 @@ func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifie return false } + // Single-flight: this slot owns the wallets end to end (gas top-up, + // corrective rebalances, triangle runs). The reaper TryLocks and skips + // its corrective action while this is held. + execMu.Lock() + defer execMu.Unlock() + ladder := downgradeLadder(tier) blockedReason := "" + // ONE corrective-transfer budget for the whole scheduler slot, shared by + // every ladder rung. Without sharing, each rung consumed its own attempt + // counter and a single slot could broadcast up to six transfers. + budget := newSlotBudget() + for i, amount := range ladder { balances, degraded, err := bc.GetAllBalancesDetailed() if err != nil || degraded { @@ -444,7 +459,7 @@ func runTierIfViable(executor *Executor, bc *BalanceChecker, slack *SlackNotifie sim := SimulateTriangleCycle(balances, amount) if !sim.Viable && rebalancer != nil { - sim, _ = rebalancer.TryUnblockTier(sim, balances, tierLabel) + sim, _ = rebalancer.TryUnblockTier(sim, balances, tierLabel, budget) } if !sim.Viable { if blockedReason == "" { diff --git a/harnesses/bridge-monitor/cmd/monitor/metrics.go b/harnesses/bridge-monitor/cmd/monitor/metrics.go index 5f8ec776..bd42f6c5 100644 --- a/harnesses/bridge-monitor/cmd/monitor/metrics.go +++ b/harnesses/bridge-monitor/cmd/monitor/metrics.go @@ -137,10 +137,11 @@ var ( Help: "1 if wallet balances are currently unreadable via both API and RPC, 0 otherwise", }) - // Auto-rebalance attempts by outcome: attempted, succeeded, failed, capped. + // Auto-rebalance attempts by outcome: attempted, succeeded, failed, + // capped, in_flight (broadcast whose bridge status never resolved). bridgeRebalanceAttempts = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "bridge_rebalance_attempts_total", - Help: "Total automatic rebalance attempts by outcome (attempted, succeeded, failed, capped)", + Help: "Total automatic rebalance attempts by outcome (attempted, succeeded, failed, capped, in_flight)", }, []string{"outcome"}) // Set to 1 when a scheduled tier was downgraded to a smaller amount because @@ -171,7 +172,7 @@ var ( // query, so the series exist on /metrics from process start instead of only // after the first event. func initSelfHealingMetrics() { - for _, outcome := range []string{"attempted", "succeeded", "failed", "capped"} { + for _, outcome := range []string{"attempted", "succeeded", "failed", "capped", "in_flight"} { bridgeRebalanceAttempts.WithLabelValues(outcome).Add(0) } bridgeTierDowngraded.WithLabelValues("300", "50").Set(0) diff --git a/harnesses/bridge-monitor/cmd/monitor/reaper.go b/harnesses/bridge-monitor/cmd/monitor/reaper.go index d40f5f6a..eceeef48 100644 --- a/harnesses/bridge-monitor/cmd/monitor/reaper.go +++ b/harnesses/bridge-monitor/cmd/monitor/reaper.go @@ -191,14 +191,24 @@ func reaperTick(fetch func() (map[string]map[string]float64, bool, error), rebal return } - if rebalancer.executor.config.DailySpentUSD >= rebalancer.executor.config.MaxDailySpendUSD { + if spent := rebalancer.executor.DailySpent(); spent >= rebalancer.executor.config.MaxDailySpendUSD { _ = slack.NotifyReaper(fmt.Sprintf( "$%.2f of %s stranded on %s for %.1fh, but daily spend limit is reached ($%.2f / $%.2f). Will retry next tick.", amount, worst.Token, worst.Chain, worstHours, - rebalancer.executor.config.DailySpentUSD, rebalancer.executor.config.MaxDailySpendUSD)) + spent, rebalancer.executor.config.MaxDailySpendUSD)) return } + // Single-flight: never broadcast while a scheduler slot (triangle run, + // rebalance, gas top-up) holds the execution lock. Skip instead of + // queueing: waiting here could fire a stale corrective transfer right + // after the slot already rebalanced the same leg. + if !execMu.TryLock() { + log.Printf("🧹 Reaper: execution lock busy (another actor is broadcasting), skipping corrective transfer this tick") + return + } + defer execMu.Unlock() + dest := neediestHomeLeg(balances) if amount > reaperMaxTransferUSD { amount = reaperMaxTransferUSD @@ -219,19 +229,24 @@ func reaperTick(fetch func() (map[string]map[string]float64, bool, error), rebal "Moving $%.2f of stranded %s from %s (stuck %.1fh) home to %s %s via cheapest bridge.", amount, worst.Token, worst.Chain, worstHours, dest.Chain, dest.Token)) + // ExecuteCheapest books the spend (realized fee on success, flat estimate + // for a broadcast that never confirmed), so no accounting here. result := rebalancer.ExecuteCheapest(route, amount) - if result == nil || !result.Success { + switch classifyCorrectiveResult(result) { + case outcomeSuccess: + _ = slack.NotifyReaper(fmt.Sprintf( + "Corrective transfer succeeded via %s (fee $%.4f, tx %s).", result.Bridge, result.ActualFeeUSD, result.TxHash)) + case outcomeInFlight: + _ = slack.NotifyReaper(fmt.Sprintf( + "Corrective transfer broadcast (tx %s) but its bridge status did not resolve. Funds may still be in flight; standing down, next tick re-evaluates fresh balances.", + result.TxHash)) + default: detail := "no bridge produced a usable quote" if result != nil && result.Error != nil { detail = result.Error.Error() } - _ = slack.NotifyReaper(fmt.Sprintf("Corrective transfer FAILED: %s. Will retry next tick.", detail)) - return + _ = slack.NotifyReaper(fmt.Sprintf("Corrective transfer FAILED before broadcast: %s. Will retry next tick.", detail)) } - - rebalancer.executor.config.DailySpentUSD += result.ActualFeeUSD - _ = slack.NotifyReaper(fmt.Sprintf( - "Corrective transfer succeeded via %s (fee $%.4f, tx %s).", result.Bridge, result.ActualFeeUSD, result.TxHash)) } // neediestHomeLeg returns the triangle leg with the lowest balance: stranded diff --git a/harnesses/bridge-monitor/cmd/monitor/rebalancer.go b/harnesses/bridge-monitor/cmd/monitor/rebalancer.go index 27d815ba..9ead71dc 100644 --- a/harnesses/bridge-monitor/cmd/monitor/rebalancer.go +++ b/harnesses/bridge-monitor/cmd/monitor/rebalancer.go @@ -14,15 +14,75 @@ import ( // like a benchmark execution. cmd/rebalance stays as the manual fallback. const ( - // Two attempts per tier slot: one transfer plus one retry covers transient - // quote failures without letting a broken bridge drain the daily budget. - maxRebalanceAttemptsPerSlot = 2 + // Two corrective transfers per SCHEDULER SLOT, shared across every rung + // of the downgrade ladder. One transfer plus one retry covers transient + // pre-broadcast failures without letting a broken bridge drain the daily + // budget. Before this was per rung, a single $300 slot could broadcast + // up to six transfers back to back. + maxCorrectiveTransfersPerSlot = 2 // Move 10 percent more than the computed shortfall so bridge fees and // slippage on the transfer itself do not leave the leg short again. rebalanceBufferFactor = 1.10 ) +// slotBudget caps corrective transfers for one scheduler slot. A single +// instance is created per slot and threaded through every ladder rung. +type slotBudget struct { + remaining int + // inFlight marks that a corrective transfer broadcast a TX whose bridge + // status never resolved. The funds are most likely still moving (legit + // fills can take 2-5 minutes, longer than our status polling), so any + // rung that needs the same destination leg must stand down for the rest + // of the slot instead of double-sending. + inFlight bool + inFlightChain string +} + +func newSlotBudget() *slotBudget { + return &slotBudget{remaining: maxCorrectiveTransfersPerSlot} +} + +// blockedByInFlight reports whether a rung needing refillChain must stand +// down because an earlier transfer to that leg is still unresolved. +func (b *slotBudget) blockedByInFlight(refillChain string) bool { + return b != nil && b.inFlight && strings.EqualFold(b.inFlightChain, refillChain) +} + +// correctiveOutcome classifies a corrective transfer attempt for retry logic. +type correctiveOutcome int + +const ( + // outcomePreBroadcast: nothing left the wallet (quote failed, approval + // failed to broadcast, insufficient funds). Safe to consume another + // attempt from the slot budget. + outcomePreBroadcast correctiveOutcome = iota + // outcomeInFlight: a TX was broadcast but the attempt did not confirm as + // a success. The deposit may well have confirmed with the bridge fill + // still pending, so the funds must be assumed to be in flight. TERMINAL + // for the slot: retrying here is exactly how a double-send happens. + outcomeInFlight + // outcomeSuccess: the transfer confirmed end to end. + outcomeSuccess +) + +// classifyCorrectiveResult decides whether a corrective transfer attempt may +// be retried. Anything that produced a TxHash moved, or may have moved, real +// funds and is terminal. Only failures that provably never broadcast are +// safe to retry. +func classifyCorrectiveResult(result *ExecutionResult) correctiveOutcome { + if result == nil { + return outcomePreBroadcast + } + if result.Success { + return outcomeSuccess + } + if result.TxHash != "" { + return outcomeInFlight + } + return outcomePreBroadcast +} + // legSpec describes one home leg of the USDC triangle. type legSpec struct { Chain string @@ -142,22 +202,37 @@ func (r *Rebalancer) canAct() bool { return r != nil && r.executor != nil && r.executor.txExecutor != nil && r.executor.txExecutor.CanExecute() } -// TryUnblockTier attempts up to maxRebalanceAttemptsPerSlot corrective -// transfers to make the given failed simulation viable. Returns the latest -// simulation and whether the tier can now run. Callers must only pass -// non-degraded balances: rebalancing on unreadable balances could move real -// funds based on phantom zeros. -func (r *Rebalancer) TryUnblockTier(sim CycleSimulation, balances map[string]map[string]float64, tierLabel string) (CycleSimulation, bool) { +// TryUnblockTier attempts corrective transfers to make the given failed +// simulation viable, consuming from the slot-wide budget shared across all +// ladder rungs. Returns the latest simulation and whether the tier can now +// run. Callers must only pass non-degraded balances: rebalancing on +// unreadable balances could move real funds based on phantom zeros. +func (r *Rebalancer) TryUnblockTier(sim CycleSimulation, balances map[string]map[string]float64, tierLabel string, budget *slotBudget) (CycleSimulation, bool) { if !r.canAct() { return sim, false } + if budget == nil { + budget = newSlotBudget() + } - for attempt := 1; attempt <= maxRebalanceAttemptsPerSlot; attempt++ { - if r.executor.config.DailySpentUSD >= r.executor.config.MaxDailySpendUSD { + for { + if budget.blockedByInFlight(sim.RefillChain) { + log.Printf("🔧 Tier $%.0f needs %s but a corrective transfer to that leg is already in flight, standing down for this slot", + sim.Tier, sim.RefillChain) + return sim, false + } + if budget.remaining <= 0 { + bridgeRebalanceAttempts.WithLabelValues("capped").Inc() + _ = r.slack.NotifyRebalance("capped", fmt.Sprintf( + "Tier $%.0f (%s) still blocked (%s) but this slot's corrective transfer budget (%d) is spent. Standing down until next slot.", + sim.Tier, tierLabel, sim.Reason, maxCorrectiveTransfersPerSlot)) + return sim, false + } + if spent := r.executor.DailySpent(); spent >= r.executor.config.MaxDailySpendUSD { bridgeRebalanceAttempts.WithLabelValues("capped").Inc() _ = r.slack.NotifyRebalance("capped", fmt.Sprintf( "Tier $%.0f (%s) blocked (%s) but daily spend limit is reached ($%.2f / $%.2f). No rebalance attempted.", - sim.Tier, tierLabel, sim.Reason, r.executor.config.DailySpentUSD, r.executor.config.MaxDailySpendUSD)) + sim.Tier, tierLabel, sim.Reason, spent, r.executor.config.MaxDailySpendUSD)) return sim, false } @@ -171,26 +246,45 @@ func (r *Rebalancer) TryUnblockTier(sim CycleSimulation, balances map[string]map return sim, false } + budget.remaining-- + attempt := maxCorrectiveTransfersPerSlot - budget.remaining bridgeRebalanceAttempts.WithLabelValues("attempted").Inc() - log.Printf("🔧 Rebalance attempt %d/%d: $%.2f %s -> %s (unblocks tier $%.0f)", - attempt, maxRebalanceAttemptsPerSlot, amountUSD, route.FromChain, route.ToChain, sim.Tier) + log.Printf("🔧 Rebalance attempt %d/%d (slot budget): $%.2f %s -> %s (unblocks tier $%.0f)", + attempt, maxCorrectiveTransfersPerSlot, amountUSD, route.FromChain, route.ToChain, sim.Tier) _ = r.slack.NotifyRebalance("attempted", fmt.Sprintf( - "Tier $%.0f (%s) blocked: %s\nMoving $%.2f from %s %s to %s %s (attempt %d/%d).", + "Tier $%.0f (%s) blocked: %s\nMoving $%.2f from %s %s to %s %s (slot attempt %d/%d).", sim.Tier, tierLabel, sim.Reason, amountUSD, route.FromChain, sourceSymbol(route), route.ToChain, sim.RefillToken, - attempt, maxRebalanceAttemptsPerSlot)) + attempt, maxCorrectiveTransfersPerSlot)) result := r.ExecuteCheapest(route, amountUSD) - if result == nil || !result.Success { + switch classifyCorrectiveResult(result) { + case outcomeInFlight: + // The deposit TX exists on-chain but the bridge status never + // resolved. Funds are most likely still moving toward the short + // leg: any further transfer to that leg this slot would be a + // double send. Terminal for the whole scheduler slot. + budget.inFlight = true + budget.inFlightChain = sim.RefillChain + bridgeRebalanceAttempts.WithLabelValues("in_flight").Inc() + _ = r.slack.NotifyRebalance("in_flight", fmt.Sprintf( + "Corrective transfer for tier $%.0f broadcast (tx %s) but its bridge status did not resolve. Funds are likely still in flight: standing down until next slot to avoid a double send.", + sim.Tier, result.TxHash)) + return sim, false + + case outcomePreBroadcast: bridgeRebalanceAttempts.WithLabelValues("failed").Inc() detail := "no bridge produced a usable quote" if result != nil && result.Error != nil { detail = result.Error.Error() } _ = r.slack.NotifyRebalance("failed", fmt.Sprintf( - "Rebalance transfer for tier $%.0f failed: %s", sim.Tier, detail)) - } else { + "Rebalance transfer for tier $%.0f failed before broadcast: %s", sim.Tier, detail)) + // Nothing left the wallet, so another attempt is safe if the + // slot budget still allows one. + continue + + case outcomeSuccess: bridgeRebalanceAttempts.WithLabelValues("succeeded").Inc() - r.executor.config.DailySpentUSD += result.ActualFeeUSD _ = r.slack.NotifyRebalance("succeeded", fmt.Sprintf( "Moved $%.2f from %s to %s via %s (fee $%.4f, tx %s). Re-checking tier $%.0f viability.", amountUSD, route.FromChain, route.ToChain, result.Bridge, result.ActualFeeUSD, result.TxHash, sim.Tier)) @@ -211,12 +305,6 @@ func (r *Rebalancer) TryUnblockTier(sim CycleSimulation, balances map[string]map return sim, true } } - - bridgeRebalanceAttempts.WithLabelValues("capped").Inc() - _ = r.slack.NotifyRebalance("capped", fmt.Sprintf( - "Tier $%.0f still not viable after %d rebalance attempts: %s", - sim.Tier, maxRebalanceAttemptsPerSlot, sim.Reason)) - return sim, false } // ExecuteCheapest quotes all three executing bridges and broadcasts through @@ -231,7 +319,12 @@ func (r *Rebalancer) ExecuteCheapest(route TestRoute, amountUSD float64) *Execut log.Printf("🔧 Cheapest bridge for %s: %s (quoted fee $%.4f)", route.Name, bridge, fee) rawUnits := toRawUnits(amountUSD) - return r.executor.executeOnBridge(bridge, route, amountUSD, amountUSD, rawUnits) + result := r.executor.executeOnBridge(bridge, route, amountUSD, amountUSD, rawUnits) + // Book the cost here so both callers (tier rebalancer and reaper) share + // the same accounting, including the flat estimate for broadcasts that + // never confirmed. + r.executor.accountSpend(result) + return result } // quoteCheapest fetches quotes from Mobula, Relay and LI.FI and returns the diff --git a/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go index 1731bb75..346c7f96 100644 --- a/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go +++ b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go @@ -1,6 +1,7 @@ package main import ( + "fmt" "math" "testing" "time" @@ -120,6 +121,74 @@ func TestDowngradeLadder(t *testing.T) { } } +func TestClassifyCorrectiveResult(t *testing.T) { + if got := classifyCorrectiveResult(nil); got != outcomePreBroadcast { + t.Errorf("nil result: got %v, want pre-broadcast", got) + } + if got := classifyCorrectiveResult(&ExecutionResult{Success: true, TxHash: "0xabc"}); got != outcomeSuccess { + t.Errorf("confirmed fill: got %v, want success", got) + } + // A broadcast whose status never resolved is TERMINAL: the deposit may + // have confirmed with the fill still pending, so a retry double-sends. + if got := classifyCorrectiveResult(&ExecutionResult{Success: false, TxHash: "0xabc", Error: fmt.Errorf("status poll failed: timeout")}); got != outcomeInFlight { + t.Errorf("status timeout with TxHash: got %v, want in-flight", got) + } + // A revert or refund also carries a hash and stays terminal for the slot. + if got := classifyCorrectiveResult(&ExecutionResult{Reverted: true, TxHash: "0xabc"}); got != outcomeInFlight { + t.Errorf("reverted with TxHash: got %v, want in-flight", got) + } + // Pre-broadcast failures (quote failed, approval failed to broadcast, + // insufficient funds) have no hash and may consume another attempt. + if got := classifyCorrectiveResult(&ExecutionResult{Error: fmt.Errorf("quote failed")}); got != outcomePreBroadcast { + t.Errorf("quote failure: got %v, want pre-broadcast", got) + } +} + +func TestSlotBudgetSharedAcrossRungs(t *testing.T) { + if maxCorrectiveTransfersPerSlot != 2 { + t.Fatalf("slot budget must be 2 corrective transfers, got %d", maxCorrectiveTransfersPerSlot) + } + b := newSlotBudget() + if b.remaining != 2 { + t.Fatalf("fresh budget: got %d remaining, want 2", b.remaining) + } + + // Rung 1 ($300) consumes one attempt, rung 2 ($50) consumes the second: + // the SAME budget instance is threaded through the ladder, so rung 3 + // ($5) has nothing left. Per-rung counters allowed up to 6 transfers. + b.remaining-- + b.remaining-- + if b.remaining > 0 { + t.Fatalf("after two attempts across rungs the slot budget must be exhausted, got %d", b.remaining) + } +} + +func TestSlotBudgetInFlightBlocksSameLeg(t *testing.T) { + b := newSlotBudget() + if b.blockedByInFlight("Solana") { + t.Fatal("fresh budget must not block any leg") + } + + // A transfer to Solana broadcast but never resolved: every later rung + // needing Solana stands down, regardless of remaining budget. + b.inFlight = true + b.inFlightChain = "Solana" + if !b.blockedByInFlight("Solana") { + t.Error("rung needing the in-flight leg must stand down") + } + if !b.blockedByInFlight("solana") { + t.Error("leg matching must be case-insensitive") + } + if b.blockedByInFlight("Base") { + t.Error("a rung needing a different leg is not blocked by the in-flight transfer") + } + + var nilBudget *slotBudget + if nilBudget.blockedByInFlight("Solana") { + t.Error("nil budget must not block") + } +} + func TestStrandedHoursComputation(t *testing.T) { tracker := NewStrandedTracker() t0 := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC) From 19357b4cb5591c76bdf5134519994cce6ae0e42f Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 16:05:08 +0200 Subject: [PATCH 12/13] bridge-monitor: SIMULATE_BALANCES only honored in dry-run mode The reaper (and any other consumer) used to accept simulated balances even in production with broadcast-capable keys, so phantom numbers could green-light real transfers. SimulateBalances now takes the execution mode and returns nil outside dry-run; startup logs a warning when the flag is set in a broadcast-capable mode. --- .../bridge-monitor/cmd/monitor/balance.go | 10 ++++++++-- .../bridge-monitor/cmd/monitor/executor.go | 2 +- harnesses/bridge-monitor/cmd/monitor/main.go | 7 +++++++ harnesses/bridge-monitor/cmd/monitor/reaper.go | 7 +++++-- .../cmd/monitor/selfheal_test.go | 18 ++++++++++++++++++ 5 files changed, 39 insertions(+), 5 deletions(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/balance.go b/harnesses/bridge-monitor/cmd/monitor/balance.go index 3d5c401b..4067fe88 100644 --- a/harnesses/bridge-monitor/cmd/monitor/balance.go +++ b/harnesses/bridge-monitor/cmd/monitor/balance.go @@ -410,12 +410,18 @@ func (bc *BalanceChecker) GetTotalBalanceUSD() (float64, error) { return total, nil } -// SimulateBalances returns fake balances for dry-run mode -func SimulateBalances() map[string]map[string]float64 { +// SimulateBalances returns fake balances for dry-run mode. Outside dry-run it +// always returns nil: simulated balances feeding a broadcast-capable process +// could green-light real transfers based on made-up numbers. +func SimulateBalances(mode ExecutionMode) map[string]map[string]float64 { // Check if we should simulate if os.Getenv("SIMULATE_BALANCES") != "true" { return nil } + if mode != ModeDryRun { + log.Printf("⚠️ SIMULATE_BALANCES=true ignored: EXECUTION_MODE is %q, simulated balances are only honored in dry-run", mode) + return nil + } log.Println("🧪 Using simulated balances (SIMULATE_BALANCES=true)") return map[string]map[string]float64{ diff --git a/harnesses/bridge-monitor/cmd/monitor/executor.go b/harnesses/bridge-monitor/cmd/monitor/executor.go index ca0bd514..a3e4052a 100644 --- a/harnesses/bridge-monitor/cmd/monitor/executor.go +++ b/harnesses/bridge-monitor/cmd/monitor/executor.go @@ -177,7 +177,7 @@ func (e *Executor) RunDryRun(route TestRoute, amountUSD float64) *ExecutionResul result.Error = err return result } - } else if balances = SimulateBalances(); balances == nil { + } else if balances = SimulateBalances(e.config.Mode); balances == nil { log.Printf(" ❌ No balance checker and SIMULATE_BALANCES not set") result.Error = fmt.Errorf("no balance source in dry-run") return result diff --git a/harnesses/bridge-monitor/cmd/monitor/main.go b/harnesses/bridge-monitor/cmd/monitor/main.go index 9ed5b3b8..a8fe5f4e 100644 --- a/harnesses/bridge-monitor/cmd/monitor/main.go +++ b/harnesses/bridge-monitor/cmd/monitor/main.go @@ -31,6 +31,13 @@ func main() { // Log configuration config.LogConfig() + // SIMULATE_BALANCES is a dry-run testing aid only. In any mode that can + // broadcast, phantom balances could green-light real transfers, so the + // flag is ignored everywhere outside dry-run (see SimulateBalances). + if config.SimulateBalances && config.ExecutionMode != string(ModeDryRun) { + log.Printf("⚠️ SIMULATE_BALANCES=true is set but EXECUTION_MODE is %q: simulated balances are IGNORED outside dry-run", config.ExecutionMode) + } + // Initialize bridges var mobulaBridge *MobulaBridge if config.MobulaAPIKey != "" { diff --git a/harnesses/bridge-monitor/cmd/monitor/reaper.go b/harnesses/bridge-monitor/cmd/monitor/reaper.go index eceeef48..a4387451 100644 --- a/harnesses/bridge-monitor/cmd/monitor/reaper.go +++ b/harnesses/bridge-monitor/cmd/monitor/reaper.go @@ -105,13 +105,16 @@ func (t *StrandedTracker) Update(balances map[string]map[string]float64, now tim func StartReaper(bc *BalanceChecker, rebalancer *Rebalancer, slack *SlackNotifier, mode string, paused bool) { // Balance source: real checker in normal operation, the SIMULATE_BALANCES // snapshot in keyless dry-run so the gauge path stays testable locally. + // SimulateBalances itself refuses to return anything outside dry-run, so + // a production process with keys can never act on made-up numbers. + execMode := ExecutionMode(mode) var fetch func() (map[string]map[string]float64, bool, error) switch { case bc != nil: fetch = bc.GetAllBalancesDetailed - case SimulateBalances() != nil: + case SimulateBalances(execMode) != nil: fetch = func() (map[string]map[string]float64, bool, error) { - return SimulateBalances(), false, nil + return SimulateBalances(execMode), false, nil } default: log.Println("🧹 Reaper disabled: no balance checker") diff --git a/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go index 346c7f96..7eb6d013 100644 --- a/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go +++ b/harnesses/bridge-monitor/cmd/monitor/selfheal_test.go @@ -189,6 +189,24 @@ func TestSlotBudgetInFlightBlocksSameLeg(t *testing.T) { } } +func TestSimulateBalancesOnlyInDryRun(t *testing.T) { + t.Setenv("SIMULATE_BALANCES", "true") + if SimulateBalances(ModeDryRun) == nil { + t.Fatal("dry-run with SIMULATE_BALANCES=true must return the snapshot") + } + if SimulateBalances(ModeProduction) != nil { + t.Fatal("production mode must ignore SIMULATE_BALANCES") + } + if SimulateBalances(ModeSingleTest) != nil { + t.Fatal("single-test mode must ignore SIMULATE_BALANCES") + } + + t.Setenv("SIMULATE_BALANCES", "false") + if SimulateBalances(ModeDryRun) != nil { + t.Fatal("no snapshot when SIMULATE_BALANCES is off") + } +} + func TestStrandedHoursComputation(t *testing.T) { tracker := NewStrandedTracker() t0 := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC) From 1b6841796ceadfcbdeb2d0110bbca39f185de14b Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Mon, 13 Jul 2026 16:16:24 +0200 Subject: [PATCH 13/13] tx executor: preserve hash on ambiguous send errors, split dry-run spend state --- .../cmd/monitor/spend_tracker.go | 7 +++++- .../bridge-monitor/cmd/monitor/tx_executor.go | 23 +++++++++++++++---- 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go b/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go index 7a885299..93865dd9 100644 --- a/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go +++ b/harnesses/bridge-monitor/cmd/monitor/spend_tracker.go @@ -113,7 +113,12 @@ func spendStatePath() string { if p := strings.TrimSpace(os.Getenv("SPEND_STATE_PATH")); p != "" { return p } - return "./spend-state.json" + // Dry-run gets its own default file so simulated spend never eats the + // production budget when both run on the same host (review pass 2). + if strings.EqualFold(strings.TrimSpace(os.Getenv("EXECUTION_MODE")), "production") { + return "./spend-state.json" + } + return "./spend-state.dryrun.json" } // failedTxFeeEstimateUSD is booked against the daily cap for any broadcast diff --git a/harnesses/bridge-monitor/cmd/monitor/tx_executor.go b/harnesses/bridge-monitor/cmd/monitor/tx_executor.go index 0e51c1bf..97aba65c 100644 --- a/harnesses/bridge-monitor/cmd/monitor/tx_executor.go +++ b/harnesses/bridge-monitor/cmd/monitor/tx_executor.go @@ -158,9 +158,14 @@ func (tx *TxExecutor) ExecuteSolanaTransaction(serializedTxBase64 string) (strin return "", fmt.Errorf("failed to sign tx: %w", err) } - // Send transaction + // Send transaction. Same send-ambiguity rule as the EVM path: a send + // error can follow node acceptance, so surface the signature when the + // client returns one and let callers treat it as in-flight. sig, err := tx.solanaClient.SendTransaction(ctx, transaction) if err != nil { + if sig != (solana.Signature{}) { + return sig.String(), fmt.Errorf("failed to send tx (may be accepted, sig %s): %w", sig.String(), err) + } return "", fmt.Errorf("failed to send tx: %w", err) } @@ -235,9 +240,14 @@ func (tx *TxExecutor) ExecuteSolanaFromInstructions(instructions []RelaySolanaIn return "", fmt.Errorf("failed to sign tx: %w", err) } - // Send transaction + // Send transaction. Same send-ambiguity rule as the EVM path: a send + // error can follow node acceptance, so surface the signature when the + // client returns one and let callers treat it as in-flight. sig, err := tx.solanaClient.SendTransaction(ctx, transaction) if err != nil { + if sig != (solana.Signature{}) { + return sig.String(), fmt.Errorf("failed to send tx (may be accepted, sig %s): %w", sig.String(), err) + } return "", fmt.Errorf("failed to send tx: %w", err) } @@ -314,13 +324,16 @@ func (tx *TxExecutor) ExecuteEVMTransaction(chain string, to string, data string return "", fmt.Errorf("failed to sign tx: %w", err) } - // Send transaction + // Send transaction. On error the node may STILL have accepted the tx + // (RPC timeout after acceptance): return the locally computed hash so + // callers classify this as in-flight, never as retry-safe. Discarding + // it caused the residual double-send hole found in review pass 2. + txHash := signedTx.Hash().Hex() err = client.SendTransaction(ctx, signedTx) if err != nil { - return "", fmt.Errorf("failed to send tx: %w", err) + return txHash, fmt.Errorf("failed to send tx (may be accepted, hash %s): %w", txHash, err) } - txHash := signedTx.Hash().Hex() log.Printf("📤 %s TX sent: %s", chain, txHash) return txHash, nil }