From c6e8bea8a95928ab79755f605b51feb38653803b Mon Sep 17 00:00:00 2001 From: nambrot Date: Tue, 3 Feb 2026 19:48:01 -0500 Subject: [PATCH 01/12] feat(rebalancer-sim): Add mock inflight context adapter for ProductionRebalancer Adds IInflightContextAdapter interface to rebalancer package and MockInflightContextAdapter to rebalancer-sim, enabling simulation of inflight context tracking without requiring a real ActionTracker/ExplorerClient. Changes: - Add IInflightContextAdapter interface to tracking module - Add optional inflightContextAdapter to RebalancerServiceConfig - Add executeWithoutTracking() fallback when ActionTracker unavailable - Create MockInflightContextAdapter maintaining pending rebalances/transfers - Wire SimulationEngine events to mock adapter callbacks - Update tests with assertions for inflight-guard and blocked-user-transfer scenarios Test results: - inflight-guard: ProductionRebalancer uses 1 rebalance vs SimpleRebalancer's 6 (83% reduction) - blocked-user-transfer: ProductionRebalancer achieves 100% completion vs SimpleRebalancer's 0% Co-Authored-By: Claude Opus 4.5 --- .../rebalancer-sim/src/SimulationEngine.ts | 39 +++++ .../src/runners/MockInflightContextAdapter.ts | 147 ++++++++++++++++++ .../src/runners/ProductionRebalancerRunner.ts | 51 +++++- typescript/rebalancer-sim/src/types.ts | 42 +++++ .../test/integration/full-simulation.test.ts | 58 +++++-- .../rebalancer/src/core/RebalancerService.ts | 114 ++++++++++++-- typescript/rebalancer/src/index.ts | 6 +- .../src/tracking/InflightContextAdapter.ts | 11 +- typescript/rebalancer/src/tracking/index.ts | 5 +- 9 files changed, 441 insertions(+), 32 deletions(-) create mode 100644 typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts diff --git a/typescript/rebalancer-sim/src/SimulationEngine.ts b/typescript/rebalancer-sim/src/SimulationEngine.ts index 78194b8e775..678c7861f56 100644 --- a/typescript/rebalancer-sim/src/SimulationEngine.ts +++ b/typescript/rebalancer-sim/src/SimulationEngine.ts @@ -13,6 +13,7 @@ import { MessageTracker } from './MessageTracker.js'; import type { BridgeMockConfig, IRebalancerRunner, + InflightContextCallbacks, MultiDomainDeploymentResult, RebalancerSimConfig, SimulationResult, @@ -43,6 +44,7 @@ export class SimulationEngine { private isRunning = false; private mailboxProcessingInterval?: NodeJS.Timeout; private mailboxProcessingInFlight = false; + private inflightCallbacks?: InflightContextCallbacks; constructor(private readonly deployment: MultiDomainDeploymentResult) { this.provider = new ethers.providers.JsonRpcProvider(deployment.anvilRpc); @@ -94,6 +96,12 @@ export class SimulationEngine { // Wire up MessageTracker events for KPI tracking this.messageTracker.on('message_delivered', (message) => { this.kpiCollector!.recordTransferComplete(message.transferId); + + // Notify inflight context adapter + this.inflightCallbacks?.onTransferDelivered( + message.origin, + message.destination, + ); }); this.messageTracker.on('message_failed', ({ message }) => { @@ -119,14 +127,33 @@ export class SimulationEngine { ); // Link bridge transfer ID to rebalance ID for completion tracking this.kpiCollector!.linkBridgeTransfer(event.transfer.id, rebalanceId); + + // Notify inflight context adapter + this.inflightCallbacks?.onRebalanceInitiated( + event.transfer.origin, + event.transfer.destination, + event.transfer.amount, + ); }); this.bridgeController.on('transfer_delivered', (event) => { this.kpiCollector!.recordRebalanceComplete(event.transfer.id); + + // Notify inflight context adapter + this.inflightCallbacks?.onRebalanceDelivered( + event.transfer.origin, + event.transfer.destination, + ); }); this.bridgeController.on('transfer_failed', (event) => { this.kpiCollector!.recordRebalanceFailed(event.transfer.id); + + // Notify inflight context adapter (treat failed as delivered for inflight tracking) + this.inflightCallbacks?.onRebalanceDelivered( + event.transfer.origin, + event.transfer.destination, + ); }); // Build warp config for rebalancer @@ -142,6 +169,9 @@ export class SimulationEngine { await rebalancer.initialize(rebalancerConfig); + // Get inflight callbacks if the rebalancer supports them + this.inflightCallbacks = rebalancer.getInflightCallbacks?.(); + // Start rebalancer daemon await rebalancer.start(); @@ -205,6 +235,8 @@ export class SimulationEngine { this.provider.removeAllListeners(); // Force polling to stop this.provider.polling = false; + // Clear inflight callbacks + this.inflightCallbacks = undefined; } } @@ -294,6 +326,13 @@ export class SimulationEngine { transfer.destination, timing.userTransferDeliveryDelay, ); + + // Notify inflight context adapter that a user transfer is pending + this.inflightCallbacks?.onTransferInitiated( + transfer.origin, + transfer.destination, + transfer.amount, + ); } catch (error) { logger.error( { diff --git a/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts b/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts new file mode 100644 index 00000000000..32affb1719f --- /dev/null +++ b/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts @@ -0,0 +1,147 @@ +import type { + IInflightContextAdapter, + InflightContext, + Route, +} from '@hyperlane-xyz/rebalancer'; +import { rootLogger } from '@hyperlane-xyz/utils'; + +const logger = rootLogger.child({ module: 'MockInflightContextAdapter' }); + +/** + * Mock implementation of IInflightContextAdapter for simulation testing. + * + * This adapter maintains lists of pending rebalances and user transfers, + * allowing tests to control inflight context without needing a real + * ActionTracker or ExplorerClient. + * + * Usage: + * - Call addPendingRebalance() when a rebalance is initiated + * - Call addPendingTransfer() when a user transfer is initiated + * - Call removePendingRebalance() when a rebalance completes + * - Call removePendingTransfer() when a user transfer completes + * - The RebalancerService calls getInflightContext() during each poll cycle + */ +export class MockInflightContextAdapter implements IInflightContextAdapter { + private pendingRebalances: Route[] = []; + private pendingTransfers: Route[] = []; + + /** + * Add a pending rebalance (called when bridge transfer is initiated) + */ + addPendingRebalance(route: Route): void { + this.pendingRebalances.push(route); + logger.debug( + { + origin: route.origin, + destination: route.destination, + amount: route.amount.toString(), + totalPending: this.pendingRebalances.length, + }, + 'Added pending rebalance', + ); + } + + /** + * Add a pending user transfer (called when user initiates warp transfer) + */ + addPendingTransfer(route: Route): void { + this.pendingTransfers.push(route); + logger.debug( + { + origin: route.origin, + destination: route.destination, + amount: route.amount.toString(), + totalPending: this.pendingTransfers.length, + }, + 'Added pending transfer', + ); + } + + /** + * Remove a pending rebalance (called when bridge transfer completes) + */ + removePendingRebalance(origin: string, destination: string): boolean { + const idx = this.pendingRebalances.findIndex( + (r) => r.origin === origin && r.destination === destination, + ); + if (idx >= 0) { + const removed = this.pendingRebalances.splice(idx, 1)[0]; + logger.debug( + { + origin, + destination, + amount: removed.amount.toString(), + remainingPending: this.pendingRebalances.length, + }, + 'Removed pending rebalance', + ); + return true; + } + logger.warn( + { origin, destination }, + 'Attempted to remove non-existent pending rebalance', + ); + return false; + } + + /** + * Remove a pending user transfer (called when warp transfer delivers) + */ + removePendingTransfer(origin: string, destination: string): boolean { + const idx = this.pendingTransfers.findIndex( + (r) => r.origin === origin && r.destination === destination, + ); + if (idx >= 0) { + const removed = this.pendingTransfers.splice(idx, 1)[0]; + logger.debug( + { + origin, + destination, + amount: removed.amount.toString(), + remainingPending: this.pendingTransfers.length, + }, + 'Removed pending transfer', + ); + return true; + } + logger.warn( + { origin, destination }, + 'Attempted to remove non-existent pending transfer', + ); + return false; + } + + /** + * Get current inflight context for strategy decision-making. + * Called by RebalancerService during each poll cycle. + */ + async getInflightContext(): Promise { + return { + pendingRebalances: [...this.pendingRebalances], + pendingTransfers: [...this.pendingTransfers], + }; + } + + /** + * Clear all pending items (useful for test cleanup) + */ + clear(): void { + this.pendingRebalances = []; + this.pendingTransfers = []; + logger.debug('Cleared all pending items'); + } + + /** + * Get count of pending rebalances + */ + getPendingRebalanceCount(): number { + return this.pendingRebalances.length; + } + + /** + * Get count of pending transfers + */ + getPendingTransferCount(): number { + return this.pendingTransfers.length; + } +} diff --git a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts index c2d0459976e..08d0e4aff43 100644 --- a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts +++ b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts @@ -11,8 +11,13 @@ import type { StrategyConfig } from '@hyperlane-xyz/rebalancer'; import { MultiProtocolProvider, MultiProvider } from '@hyperlane-xyz/sdk'; import { ProtocolType, rootLogger } from '@hyperlane-xyz/utils'; -import type { IRebalancerRunner, RebalancerSimConfig } from '../types.js'; +import type { + IRebalancerRunner, + InflightContextCallbacks, + RebalancerSimConfig, +} from '../types.js'; +import { MockInflightContextAdapter } from './MockInflightContextAdapter.js'; import { SimulationRegistry } from './SimulationRegistry.js'; // Silent logger for the rebalancer service (internal) @@ -112,12 +117,16 @@ export class ProductionRebalancerRunner private config?: RebalancerSimConfig; private service?: RebalancerService; private running = false; + private mockAdapter?: MockInflightContextAdapter; async initialize(config: RebalancerSimConfig): Promise { // Cleanup any previously running instance await cleanupProductionRebalancer(); this.config = config; + + // Create mock inflight context adapter for simulation + this.mockAdapter = new MockInflightContextAdapter(); } async start(): Promise { @@ -217,7 +226,7 @@ export class ProductionRebalancerRunner strategyConfig, ] as StrategyConfig[]); - // Create service + // Create service with mock inflight context adapter this.service = new RebalancerService( multiProvider, multiProtocolProvider, @@ -229,6 +238,7 @@ export class ProductionRebalancerRunner monitorOnly: false, withMetrics: false, logger: silentLogger, + inflightContextAdapter: this.mockAdapter, }, ); @@ -274,6 +284,10 @@ export class ProductionRebalancerRunner } this.config = undefined; + if (this.mockAdapter) { + this.mockAdapter.clear(); + this.mockAdapter = undefined; + } this.removeAllListeners(); } @@ -286,4 +300,37 @@ export class ProductionRebalancerRunner const settleTime = Math.min(timeoutMs, 2000); await new Promise((resolve) => setTimeout(resolve, settleTime)); } + + /** + * Get callbacks for wiring inflight context updates. + * SimulationEngine uses these to notify about pending transfers and rebalances. + */ + getInflightCallbacks(): InflightContextCallbacks | undefined { + if (!this.mockAdapter) { + return undefined; + } + + return { + onTransferInitiated: ( + origin: string, + destination: string, + amount: bigint, + ) => { + this.mockAdapter!.addPendingTransfer({ origin, destination, amount }); + }, + onTransferDelivered: (origin: string, destination: string) => { + this.mockAdapter!.removePendingTransfer(origin, destination); + }, + onRebalanceInitiated: ( + origin: string, + destination: string, + amount: bigint, + ) => { + this.mockAdapter!.addPendingRebalance({ origin, destination, amount }); + }, + onRebalanceDelivered: (origin: string, destination: string) => { + this.mockAdapter!.removePendingRebalance(origin, destination); + }, + }; + } } diff --git a/typescript/rebalancer-sim/src/types.ts b/typescript/rebalancer-sim/src/types.ts index de896f9c7f2..dd97d062e65 100644 --- a/typescript/rebalancer-sim/src/types.ts +++ b/typescript/rebalancer-sim/src/types.ts @@ -377,6 +377,41 @@ export interface ChainStrategyConfig { bridgeLockTime: number; } +/** + * Callbacks for wiring inflight context updates in simulation. + * The simulation engine uses these to notify the rebalancer about + * pending transfers and rebalances. + */ +export interface InflightContextCallbacks { + /** + * Called when a user transfer is initiated (before delivery) + */ + onTransferInitiated: ( + origin: string, + destination: string, + amount: bigint, + ) => void; + + /** + * Called when a user transfer is delivered + */ + onTransferDelivered: (origin: string, destination: string) => void; + + /** + * Called when a rebalance transfer is initiated via bridge + */ + onRebalanceInitiated: ( + origin: string, + destination: string, + amount: bigint, + ) => void; + + /** + * Called when a rebalance transfer completes via bridge + */ + onRebalanceDelivered: (origin: string, destination: string) => void; +} + /** * Interface for rebalancer runners in simulation */ @@ -413,6 +448,13 @@ export interface IRebalancerRunner { * Subscribe to rebalancer events */ on(event: 'rebalance', listener: (e: RebalancerEvent) => void): this; + + /** + * Get callbacks for inflight context updates (optional). + * If supported, SimulationEngine wires these to bridge/message events. + * Returns undefined if the runner doesn't support inflight tracking. + */ + getInflightCallbacks?(): InflightContextCallbacks | undefined; } /** diff --git a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts index 506f8a0b6e0..bfc24e6580d 100644 --- a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts +++ b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts @@ -255,17 +255,20 @@ describe('Rebalancer Simulation', function () { /** * Blocked User Transfer Test * - * Tests the gap when user transfers are BLOCKED at destination due to - * insufficient collateral. With 50/50 weights and 50% tolerance, state - * appears within tolerance after user initiates transfer (69%/31% is - * within 25-75% range). But destination chain2 can't pay out (40 tokens - * < 50 needed), so transfer stays stuck. + * Tests that the ProductionRebalancer proactively adds collateral when + * user transfers are pending but blocked due to insufficient collateral. * - * Neither rebalancer acts because weights are within tolerance. - * With future mock Explorer, rebalancer should see blocked transfer - * and proactively add collateral to chain2. + * Scenario: 130 total tokens split 90/40 between chain1/chain2. + * User initiates 50 token transfer from chain1 → chain2. + * chain2 only has 40 tokens but needs 50 to pay out. + * + * SimpleRebalancer: Only sees on-chain balances, doesn't know about pending + * transfer, weights appear within tolerance → no action → transfer stuck + * + * ProductionRebalancer: Mock adapter tracks pending transfer, strategy + * reserves collateral for it, detects deficit → rebalances → transfer succeeds */ - it('blocked-user-transfer: demonstrates blocked transfer without Explorer (report only)', async function () { + it('blocked-user-transfer: ProductionRebalancer proactively adds collateral for pending transfers', async function () { this.timeout(120000); const { results } = await runScenarioWithRebalancers( @@ -281,15 +284,38 @@ describe('Rebalancer Simulation', function () { ` ${result.rebalancerName}: completion=${(result.kpis.completionRate * 100).toFixed(0)}%, rebalances=${result.kpis.totalRebalances}`, ); } - console.log( - ' (Expected: 0% completion for both - weights within tolerance, no rebalancing)', - ); - console.log( - ' (Transfer blocked: chain2 has 40 tokens but needs 50 to pay out)', + + // Find results by rebalancer type + const productionResult = results.find( + (r) => r.rebalancerName === 'ProductionRebalancerService', ); - console.log( - ' (With Explorer mock: rebalancer should see blocked transfer and act)', + const simpleResult = results.find( + (r) => r.rebalancerName === 'SimpleRebalancer', ); + + // SimpleRebalancer without inflight tracking should fail to complete + if (simpleResult) { + expect(simpleResult.kpis.completionRate).to.equal( + 0, + 'SimpleRebalancer should have 0% completion (blocked transfer)', + ); + expect(simpleResult.kpis.totalRebalances).to.equal( + 0, + 'SimpleRebalancer should not rebalance (weights within tolerance)', + ); + } + + // ProductionRebalancer with inflight tracking should complete + if (productionResult) { + expect(productionResult.kpis.completionRate).to.equal( + 1.0, + 'ProductionRebalancer should have 100% completion (proactive collateral)', + ); + expect(productionResult.kpis.totalRebalances).to.be.greaterThan( + 0, + 'ProductionRebalancer should rebalance (sees pending transfer deficit)', + ); + } }); // ============================================================================ diff --git a/typescript/rebalancer/src/core/RebalancerService.ts b/typescript/rebalancer/src/core/RebalancerService.ts index 3c42d53fdc5..d06a448ace2 100644 --- a/typescript/rebalancer/src/core/RebalancerService.ts +++ b/typescript/rebalancer/src/core/RebalancerService.ts @@ -36,7 +36,7 @@ import { Metrics } from '../metrics/Metrics.js'; import { Monitor } from '../monitor/Monitor.js'; import { type IActionTracker, - InflightContextAdapter, + type IInflightContextAdapter, } from '../tracking/index.js'; import { getRawBalances } from '../utils/balanceUtils.js'; @@ -61,6 +61,13 @@ export interface RebalancerServiceConfig { /** Service version for logging */ version?: string; + + /** + * Optional pre-configured inflight context adapter. + * If provided, skips ActionTracker creation and uses this adapter directly. + * Useful for simulation/testing where inflight context is mocked externally. + */ + inflightContextAdapter?: IInflightContextAdapter; } export interface ManualRebalanceRequest { @@ -120,7 +127,7 @@ export class RebalancerService { private metrics?: Metrics; private mode: 'manual' | 'daemon'; private actionTracker?: IActionTracker; - private inflightContextAdapter?: InflightContextAdapter; + private inflightContextAdapter?: IInflightContextAdapter; constructor( private readonly multiProvider: MultiProvider, @@ -179,13 +186,22 @@ export class RebalancerService { ); } - // Create ActionTracker for tracking inflight actions - const { tracker, adapter } = - await this.contextFactory.createActionTracker(); - this.actionTracker = tracker; - this.inflightContextAdapter = adapter; - await this.actionTracker.initialize(); - this.logger.info('ActionTracker initialized'); + // Create or use provided inflight context adapter + if (this.config.inflightContextAdapter) { + // Use externally provided adapter (e.g., for simulation/testing) + this.inflightContextAdapter = this.config.inflightContextAdapter; + this.logger.info( + 'Using externally provided InflightContextAdapter (no ActionTracker)', + ); + } else { + // Create ActionTracker for tracking inflight actions + const { tracker, adapter } = + await this.contextFactory.createActionTracker(); + this.actionTracker = tracker; + this.inflightContextAdapter = adapter; + await this.actionTracker.initialize(); + this.logger.info('ActionTracker initialized'); + } this.logger.info( { @@ -418,12 +434,21 @@ export class RebalancerService { /** * Execute rebalancing with intent tracking. * Creates intents and assigns IDs to routes before execution, then processes results by ID. + * If ActionTracker is not available (e.g., using external inflightContextAdapter), + * falls back to simple execution without tracking. */ private async executeWithTracking( strategyRoutes: StrategyRoute[], ): Promise { - if (!this.rebalancer || !this.actionTracker) { - this.logger.warn('Rebalancer or ActionTracker not available, skipping'); + if (!this.rebalancer) { + this.logger.warn('Rebalancer not available, skipping'); + return; + } + + // Fall back to simple execution if ActionTracker is not available + // (e.g., when using an external inflightContextAdapter for simulation) + if (!this.actionTracker) { + await this.executeWithoutTracking(strategyRoutes); return; } @@ -528,6 +553,73 @@ export class RebalancerService { } } + /** + * Execute rebalancing without ActionTracker tracking. + * Used when an external inflightContextAdapter is provided (e.g., simulation). + * Simply executes the routes and logs results. + */ + private async executeWithoutTracking( + strategyRoutes: StrategyRoute[], + ): Promise { + assert(this.rebalancer, 'Rebalancer must be available'); + + // Generate simple IDs for routes (not tracked in ActionTracker) + const rebalanceRoutes: RebalanceRoute[] = strategyRoutes.map((route) => ({ + ...route, + intentId: randomUUID(), + })); + + this.logger.debug( + { routeCount: rebalanceRoutes.length }, + 'Executing rebalance routes (without tracking)', + ); + + try { + const results = await this.rebalancer.rebalance(rebalanceRoutes); + const failedResults = results.filter((r) => !r.success); + if (failedResults.length > 0) { + this.metrics?.recordRebalancerFailure(); + this.logger.warn( + { failureCount: failedResults.length, total: results.length }, + 'Rebalancer cycle completed with failures', + ); + } else { + this.metrics?.recordRebalancerSuccess(); + this.logger.info( + { routeCount: results.length }, + 'Rebalancer completed a cycle successfully (no tracking)', + ); + } + + // Log results for debugging + for (const result of results) { + if (result.success) { + this.logger.debug( + { + origin: result.route.origin, + destination: result.route.destination, + messageId: result.messageId, + txHash: result.txHash, + }, + 'Rebalance executed successfully', + ); + } else { + this.logger.warn( + { + origin: result.route.origin, + destination: result.route.destination, + error: result.error, + }, + 'Rebalance execution failed', + ); + } + } + } catch (error: any) { + this.metrics?.recordRebalancerFailure(); + this.logger.error({ error }, 'Error while rebalancing'); + } + } + /** * Event handler for monitor errors */ diff --git a/typescript/rebalancer/src/index.ts b/typescript/rebalancer/src/index.ts index ba0af6af488..f65300b6b65 100644 --- a/typescript/rebalancer/src/index.ts +++ b/typescript/rebalancer/src/index.ts @@ -65,6 +65,7 @@ export type { export type { IStrategy, StrategyRoute, + Route, RawBalances, InflightContext, } from './interfaces/IStrategy.js'; @@ -92,7 +93,10 @@ export { isCollateralizedTokenEligibleForRebalancing } from './utils/tokenUtils. export { ExplorerClient } from './utils/ExplorerClient.js'; // Tracking -export { InflightContextAdapter } from './tracking/InflightContextAdapter.js'; +export { + type IInflightContextAdapter, + InflightContextAdapter, +} from './tracking/InflightContextAdapter.js'; // Factory export { RebalancerContextFactory } from './factories/RebalancerContextFactory.js'; diff --git a/typescript/rebalancer/src/tracking/InflightContextAdapter.ts b/typescript/rebalancer/src/tracking/InflightContextAdapter.ts index fac6be9228c..6874d1ec69a 100644 --- a/typescript/rebalancer/src/tracking/InflightContextAdapter.ts +++ b/typescript/rebalancer/src/tracking/InflightContextAdapter.ts @@ -4,11 +4,20 @@ import type { InflightContext } from '../interfaces/IStrategy.js'; import type { IActionTracker } from './IActionTracker.js'; +/** + * Interface for adapters that provide inflight context to strategies. + * This allows mocking inflight context in simulations without needing + * a full ActionTracker/ExplorerClient setup. + */ +export interface IInflightContextAdapter { + getInflightContext(): Promise; +} + /** * Adapter that converts ActionTracker data to strategy-consumable InflightContext. * Handles conversion from Domain IDs (used by ActionTracker) to ChainNames (used by Strategy). */ -export class InflightContextAdapter { +export class InflightContextAdapter implements IInflightContextAdapter { constructor( private readonly actionTracker: IActionTracker, private readonly multiProvider: MultiProvider, diff --git a/typescript/rebalancer/src/tracking/index.ts b/typescript/rebalancer/src/tracking/index.ts index 46a295c168e..3a5b1507c54 100644 --- a/typescript/rebalancer/src/tracking/index.ts +++ b/typescript/rebalancer/src/tracking/index.ts @@ -33,4 +33,7 @@ export type { } from './IActionTracker.js'; // Export InflightContextAdapter -export { InflightContextAdapter } from './InflightContextAdapter.js'; +export { + type IInflightContextAdapter, + InflightContextAdapter, +} from './InflightContextAdapter.js'; From 96c81285d975ddb026899d8503497ea4f6e5e1c1 Mon Sep 17 00:00:00 2001 From: nambrot Date: Wed, 4 Feb 2026 13:30:41 -0500 Subject: [PATCH 02/12] fix(rebalancer-sim): Use transfer IDs for inflight tracking instead of origin/destination Addresses PR review feedback - matching pending items by origin+destination could remove the wrong item when multiple concurrent transfers exist on the same route. Now uses unique transfer IDs for correct removal. Co-Authored-By: Claude Opus 4.5 --- .../rebalancer-sim/src/SimulationEngine.ts | 17 ++---- .../src/runners/MockInflightContextAdapter.ts | 54 +++++++++---------- .../src/runners/ProductionRebalancerRunner.ts | 22 +++++--- typescript/rebalancer-sim/src/types.ts | 6 ++- 4 files changed, 49 insertions(+), 50 deletions(-) diff --git a/typescript/rebalancer-sim/src/SimulationEngine.ts b/typescript/rebalancer-sim/src/SimulationEngine.ts index 678c7861f56..231858030f3 100644 --- a/typescript/rebalancer-sim/src/SimulationEngine.ts +++ b/typescript/rebalancer-sim/src/SimulationEngine.ts @@ -98,10 +98,7 @@ export class SimulationEngine { this.kpiCollector!.recordTransferComplete(message.transferId); // Notify inflight context adapter - this.inflightCallbacks?.onTransferDelivered( - message.origin, - message.destination, - ); + this.inflightCallbacks?.onTransferDelivered(message.transferId); }); this.messageTracker.on('message_failed', ({ message }) => { @@ -130,6 +127,7 @@ export class SimulationEngine { // Notify inflight context adapter this.inflightCallbacks?.onRebalanceInitiated( + event.transfer.id, event.transfer.origin, event.transfer.destination, event.transfer.amount, @@ -140,20 +138,14 @@ export class SimulationEngine { this.kpiCollector!.recordRebalanceComplete(event.transfer.id); // Notify inflight context adapter - this.inflightCallbacks?.onRebalanceDelivered( - event.transfer.origin, - event.transfer.destination, - ); + this.inflightCallbacks?.onRebalanceDelivered(event.transfer.id); }); this.bridgeController.on('transfer_failed', (event) => { this.kpiCollector!.recordRebalanceFailed(event.transfer.id); // Notify inflight context adapter (treat failed as delivered for inflight tracking) - this.inflightCallbacks?.onRebalanceDelivered( - event.transfer.origin, - event.transfer.destination, - ); + this.inflightCallbacks?.onRebalanceDelivered(event.transfer.id); }); // Build warp config for rebalancer @@ -329,6 +321,7 @@ export class SimulationEngine { // Notify inflight context adapter that a user transfer is pending this.inflightCallbacks?.onTransferInitiated( + transfer.id, transfer.origin, transfer.destination, transfer.amount, diff --git a/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts b/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts index 32affb1719f..e09d129f858 100644 --- a/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts +++ b/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts @@ -22,16 +22,17 @@ const logger = rootLogger.child({ module: 'MockInflightContextAdapter' }); * - The RebalancerService calls getInflightContext() during each poll cycle */ export class MockInflightContextAdapter implements IInflightContextAdapter { - private pendingRebalances: Route[] = []; - private pendingTransfers: Route[] = []; + private pendingRebalances: Array<{ id: string; route: Route }> = []; + private pendingTransfers: Array<{ id: string; route: Route }> = []; /** * Add a pending rebalance (called when bridge transfer is initiated) */ - addPendingRebalance(route: Route): void { - this.pendingRebalances.push(route); + addPendingRebalance(id: string, route: Route): void { + this.pendingRebalances.push({ id, route }); logger.debug( { + id, origin: route.origin, destination: route.destination, amount: route.amount.toString(), @@ -44,10 +45,11 @@ export class MockInflightContextAdapter implements IInflightContextAdapter { /** * Add a pending user transfer (called when user initiates warp transfer) */ - addPendingTransfer(route: Route): void { - this.pendingTransfers.push(route); + addPendingTransfer(id: string, route: Route): void { + this.pendingTransfers.push({ id, route }); logger.debug( { + id, origin: route.origin, destination: route.destination, amount: route.amount.toString(), @@ -60,54 +62,46 @@ export class MockInflightContextAdapter implements IInflightContextAdapter { /** * Remove a pending rebalance (called when bridge transfer completes) */ - removePendingRebalance(origin: string, destination: string): boolean { - const idx = this.pendingRebalances.findIndex( - (r) => r.origin === origin && r.destination === destination, - ); + removePendingRebalance(id: string): boolean { + const idx = this.pendingRebalances.findIndex((r) => r.id === id); if (idx >= 0) { const removed = this.pendingRebalances.splice(idx, 1)[0]; logger.debug( { - origin, - destination, - amount: removed.amount.toString(), + id, + origin: removed.route.origin, + destination: removed.route.destination, + amount: removed.route.amount.toString(), remainingPending: this.pendingRebalances.length, }, 'Removed pending rebalance', ); return true; } - logger.warn( - { origin, destination }, - 'Attempted to remove non-existent pending rebalance', - ); + logger.warn({ id }, 'Attempted to remove non-existent pending rebalance'); return false; } /** * Remove a pending user transfer (called when warp transfer delivers) */ - removePendingTransfer(origin: string, destination: string): boolean { - const idx = this.pendingTransfers.findIndex( - (r) => r.origin === origin && r.destination === destination, - ); + removePendingTransfer(id: string): boolean { + const idx = this.pendingTransfers.findIndex((r) => r.id === id); if (idx >= 0) { const removed = this.pendingTransfers.splice(idx, 1)[0]; logger.debug( { - origin, - destination, - amount: removed.amount.toString(), + id, + origin: removed.route.origin, + destination: removed.route.destination, + amount: removed.route.amount.toString(), remainingPending: this.pendingTransfers.length, }, 'Removed pending transfer', ); return true; } - logger.warn( - { origin, destination }, - 'Attempted to remove non-existent pending transfer', - ); + logger.warn({ id }, 'Attempted to remove non-existent pending transfer'); return false; } @@ -117,8 +111,8 @@ export class MockInflightContextAdapter implements IInflightContextAdapter { */ async getInflightContext(): Promise { return { - pendingRebalances: [...this.pendingRebalances], - pendingTransfers: [...this.pendingTransfers], + pendingRebalances: this.pendingRebalances.map((r) => r.route), + pendingTransfers: this.pendingTransfers.map((r) => r.route), }; } diff --git a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts index 08d0e4aff43..102589a0471 100644 --- a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts +++ b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts @@ -312,24 +312,34 @@ export class ProductionRebalancerRunner return { onTransferInitiated: ( + id: string, origin: string, destination: string, amount: bigint, ) => { - this.mockAdapter!.addPendingTransfer({ origin, destination, amount }); + this.mockAdapter!.addPendingTransfer(id, { + origin, + destination, + amount, + }); }, - onTransferDelivered: (origin: string, destination: string) => { - this.mockAdapter!.removePendingTransfer(origin, destination); + onTransferDelivered: (id: string) => { + this.mockAdapter!.removePendingTransfer(id); }, onRebalanceInitiated: ( + id: string, origin: string, destination: string, amount: bigint, ) => { - this.mockAdapter!.addPendingRebalance({ origin, destination, amount }); + this.mockAdapter!.addPendingRebalance(id, { + origin, + destination, + amount, + }); }, - onRebalanceDelivered: (origin: string, destination: string) => { - this.mockAdapter!.removePendingRebalance(origin, destination); + onRebalanceDelivered: (id: string) => { + this.mockAdapter!.removePendingRebalance(id); }, }; } diff --git a/typescript/rebalancer-sim/src/types.ts b/typescript/rebalancer-sim/src/types.ts index dd97d062e65..9150ed682f3 100644 --- a/typescript/rebalancer-sim/src/types.ts +++ b/typescript/rebalancer-sim/src/types.ts @@ -387,6 +387,7 @@ export interface InflightContextCallbacks { * Called when a user transfer is initiated (before delivery) */ onTransferInitiated: ( + id: string, origin: string, destination: string, amount: bigint, @@ -395,12 +396,13 @@ export interface InflightContextCallbacks { /** * Called when a user transfer is delivered */ - onTransferDelivered: (origin: string, destination: string) => void; + onTransferDelivered: (id: string) => void; /** * Called when a rebalance transfer is initiated via bridge */ onRebalanceInitiated: ( + id: string, origin: string, destination: string, amount: bigint, @@ -409,7 +411,7 @@ export interface InflightContextCallbacks { /** * Called when a rebalance transfer completes via bridge */ - onRebalanceDelivered: (origin: string, destination: string) => void; + onRebalanceDelivered: (id: string) => void; } /** From dba36086261a405660864a7086ea6f680051d041 Mon Sep 17 00:00:00 2001 From: nambrot Date: Wed, 4 Feb 2026 15:38:09 -0500 Subject: [PATCH 03/12] refactor(rebalancer-sim): Use MockActionTracker instead of MockInflightContextAdapter The simulation now mocks at the ActionTracker level rather than the InflightContextAdapter level. This provides cleaner integration with RebalancerService and properly demonstrates inflight tracking benefits. Key changes: - Add MockActionTracker implementing IActionTracker interface - Remove MockInflightContextAdapter (replaced by MockActionTracker) - Update RebalancerService to accept optional actionTracker config - Handle MockValueTransferBridge not emitting Dispatch events: - failRebalanceIntent keeps intent as in_progress (bridge succeeded) - createActionForPendingIntent creates actions when bridge events fire - completeRebalanceByRoute marks actions complete on delivery Results demonstrate inflight tracking benefits: - inflight-guard: ProductionRebalancer 5 rebalances vs SimpleRebalancer 18 - blocked-user-transfer: ProductionRebalancer 100% completion vs 0% Co-Authored-By: Claude Opus 4.5 --- .../src/runners/MockActionTracker.ts | 452 ++++++++++++++++++ .../src/runners/MockInflightContextAdapter.ts | 141 ------ .../src/runners/ProductionRebalancerRunner.ts | 83 +++- .../test/integration/full-simulation.test.ts | 3 +- .../test/integration/harness-setup.test.ts | 6 +- typescript/rebalancer-sim/test/utils/anvil.ts | 40 +- .../rebalancer/src/core/RebalancerService.ts | 107 +---- typescript/rebalancer/src/index.ts | 11 +- .../src/tracking/InflightContextAdapter.ts | 11 +- typescript/rebalancer/src/tracking/index.ts | 5 +- 10 files changed, 570 insertions(+), 289 deletions(-) create mode 100644 typescript/rebalancer-sim/src/runners/MockActionTracker.ts delete mode 100644 typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts diff --git a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts new file mode 100644 index 00000000000..06d8fa24262 --- /dev/null +++ b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts @@ -0,0 +1,452 @@ +import type { + CreateRebalanceActionParams, + CreateRebalanceIntentParams, + IActionTracker, +} from '@hyperlane-xyz/rebalancer'; +import type { Address, Domain } from '@hyperlane-xyz/utils'; +import { rootLogger } from '@hyperlane-xyz/utils'; + +const logger = rootLogger.child({ module: 'MockActionTracker' }); + +/** + * Transfer record matching the real Transfer type. + */ +interface Transfer { + id: string; + origin: Domain; + destination: Domain; + amount: bigint; + status: 'in_progress' | 'complete'; + messageId: string; + sender: Address; + recipient: Address; + createdAt: number; + updatedAt: number; +} + +/** + * RebalanceIntent record matching the real type. + */ +interface RebalanceIntent { + id: string; + origin: Domain; + destination: Domain; + amount: bigint; + bridge?: Address; + status: 'not_started' | 'in_progress' | 'complete' | 'cancelled' | 'failed'; + fulfilledAmount: bigint; + priority?: number; + strategyType?: string; + createdAt: number; + updatedAt: number; +} + +/** + * RebalanceAction record matching the real type. + */ +interface RebalanceAction { + id: string; + intentId: string; + origin: Domain; + destination: Domain; + amount: bigint; + messageId: string; + txHash?: string; + status: 'in_progress' | 'complete' | 'failed'; + createdAt: number; + updatedAt: number; +} + +/** + * Mock implementation of IActionTracker for simulation testing. + * + * This tracker maintains in-memory state without requiring a real + * ExplorerClient or on-chain queries. The simulation engine updates + * this tracker via callbacks when transfers/rebalances are initiated + * or completed. + */ +export class MockActionTracker implements IActionTracker { + private transfers = new Map(); + private intents = new Map(); + private actions = new Map(); + private idCounter = 0; + + async initialize(): Promise { + logger.debug('MockActionTracker initialized'); + } + + // === Sync Operations (no-ops in simulation) === + + async syncTransfers(): Promise { + // No-op: simulation manages state directly via callbacks + } + + async syncRebalanceIntents(): Promise { + // No-op: simulation manages state directly + } + + async syncRebalanceActions(): Promise { + // No-op: simulation manages state directly + } + + // === Transfer Management === + + /** + * Add a transfer (called by simulation when user transfer is initiated). + */ + addTransfer( + id: string, + origin: Domain, + destination: Domain, + amount: bigint, + ): void { + const now = Date.now(); + this.transfers.set(id, { + id, + origin, + destination, + amount, + status: 'in_progress', + messageId: id, // Use transfer ID as message ID in simulation + sender: '0x0000000000000000000000000000000000000000', + recipient: '0x0000000000000000000000000000000000000000', + createdAt: now, + updatedAt: now, + }); + logger.debug( + { id, origin, destination, amount: amount.toString() }, + 'Added transfer', + ); + } + + /** + * Remove a transfer (called by simulation when transfer delivers or fails). + */ + removeTransfer(id: string): void { + this.transfers.delete(id); + logger.debug({ id }, 'Transfer removed'); + } + + async getInProgressTransfers(): Promise { + return Array.from(this.transfers.values()).filter( + (t) => t.status === 'in_progress', + ); + } + + async getTransfersByDestination(destination: Domain): Promise { + return Array.from(this.transfers.values()).filter( + (t) => t.destination === destination && t.status === 'in_progress', + ); + } + + // === RebalanceIntent Management === + + async createRebalanceIntent( + params: CreateRebalanceIntentParams, + ): Promise { + const id = `intent-${++this.idCounter}`; + const now = Date.now(); + const intent: RebalanceIntent = { + id, + origin: params.origin, + destination: params.destination, + amount: params.amount, + bridge: params.bridge, + priority: params.priority, + strategyType: params.strategyType, + status: 'not_started', + fulfilledAmount: 0n, + createdAt: now, + updatedAt: now, + }; + this.intents.set(id, intent); + logger.debug( + { + id, + origin: params.origin, + destination: params.destination, + amount: params.amount.toString(), + }, + 'Created rebalance intent', + ); + return intent; + } + + async getActiveRebalanceIntents(): Promise { + return Array.from(this.intents.values()).filter( + (i) => i.status === 'not_started' || i.status === 'in_progress', + ); + } + + async getRebalanceIntentsByDestination( + destination: Domain, + ): Promise { + return Array.from(this.intents.values()).filter( + (i) => + i.destination === destination && + (i.status === 'not_started' || i.status === 'in_progress'), + ); + } + + async completeRebalanceIntent(id: string): Promise { + const intent = this.intents.get(id); + if (intent) { + intent.status = 'complete'; + intent.updatedAt = Date.now(); + logger.debug({ id }, 'Rebalance intent completed'); + } + } + + async cancelRebalanceIntent(id: string): Promise { + const intent = this.intents.get(id); + if (intent) { + intent.status = 'cancelled'; + intent.updatedAt = Date.now(); + logger.debug({ id }, 'Rebalance intent cancelled'); + } + } + + async failRebalanceIntent(id: string): Promise { + const intent = this.intents.get(id); + if (intent) { + // In simulation, RebalancerService calls failRebalanceIntent when no Dispatch event + // is found (MockValueTransferBridge doesn't emit Dispatch). But the bridge transfer + // DID succeed - we'll see the SentTransferRemote event. So we keep the intent as + // in_progress to allow createActionForPendingIntent to work when the event fires. + // + // Note: This is a simulation-specific behavior. In production, no Dispatch event + // means the transfer actually failed. + intent.status = 'in_progress'; + intent.updatedAt = Date.now(); + logger.debug( + { id, originalIntentAmount: intent.amount.toString() }, + 'Rebalance intent "failed" - keeping as in_progress for simulation (bridge transfer succeeded)', + ); + } + } + + // === RebalanceAction Management === + + async createRebalanceAction( + params: CreateRebalanceActionParams, + ): Promise { + const id = `action-${++this.idCounter}`; + const now = Date.now(); + const action: RebalanceAction = { + id, + intentId: params.intentId, + origin: params.origin, + destination: params.destination, + amount: params.amount, + messageId: params.messageId, + txHash: params.txHash, + status: 'in_progress', + createdAt: now, + updatedAt: now, + }; + this.actions.set(id, action); + + // Transition parent intent to in_progress + const intent = this.intents.get(params.intentId); + if (intent && intent.status === 'not_started') { + intent.status = 'in_progress'; + intent.updatedAt = now; + } + + logger.info( + { + id, + intentId: params.intentId, + messageId: params.messageId, + origin: params.origin, + destination: params.destination, + amount: params.amount.toString(), + }, + 'Created rebalance action', + ); + return action; + } + + async completeRebalanceAction(id: string): Promise { + const action = this.actions.get(id); + if (action) { + action.status = 'complete'; + action.updatedAt = Date.now(); + + // Update parent intent's fulfilledAmount + const intent = this.intents.get(action.intentId); + if (intent) { + intent.fulfilledAmount += action.amount; + intent.updatedAt = Date.now(); + } + + logger.debug({ id }, 'Rebalance action completed'); + } + } + + async failRebalanceAction(id: string): Promise { + const action = this.actions.get(id); + if (action) { + action.status = 'failed'; + action.updatedAt = Date.now(); + logger.debug({ id }, 'Rebalance action failed'); + } + } + + // === Debug === + + async logStoreContents(): Promise { + logger.debug( + { + transfers: this.transfers.size, + intents: this.intents.size, + actions: this.actions.size, + }, + 'MockActionTracker store contents', + ); + } + + /** + * Clear all state (useful for test cleanup). + */ + clear(): void { + this.transfers.clear(); + this.intents.clear(); + this.actions.clear(); + this.idCounter = 0; + logger.debug('MockActionTracker cleared'); + } + + // === Simulation-specific methods === + + /** + * Create an action for an intent by matching origin/destination/amount. + * Used by simulation when bridge transfer is initiated, since RebalancerService + * can't extract messageId from MockValueTransferBridge (no Dispatch event). + * Returns the created action, or null if no matching intent found. + */ + createActionForPendingIntent( + origin: Domain, + destination: Domain, + amount: bigint, + bridgeTransferId: string, + ): RebalanceAction | null { + // Find matching intent that needs an action (not_started or in_progress without enough fulfilled) + const allIntents = Array.from(this.intents.values()); + const matchingIntents = allIntents + .filter( + (i) => + i.origin === origin && + i.destination === destination && + i.amount === amount && + (i.status === 'not_started' || i.status === 'in_progress'), + ) + .sort((a, b) => a.createdAt - b.createdAt); + + if (matchingIntents.length === 0) { + logger.warn( + { origin, destination, amount: amount.toString() }, + 'No matching intent found for bridge transfer', + ); + return null; + } + + const intent = matchingIntents[0]; + const now = Date.now(); + const actionId = `action-${++this.idCounter}`; + + const action: RebalanceAction = { + id: actionId, + intentId: intent.id, + origin, + destination, + amount, + messageId: bridgeTransferId, // Use bridge transfer ID as pseudo-messageId + status: 'in_progress', + createdAt: now, + updatedAt: now, + }; + + this.actions.set(actionId, action); + + // Transition intent to in_progress if needed + if (intent.status === 'not_started') { + intent.status = 'in_progress'; + intent.updatedAt = now; + } + + logger.debug( + { + actionId, + intentId: intent.id, + bridgeTransferId, + origin, + destination, + amount: amount.toString(), + }, + 'Created action for pending intent (simulation bridge transfer)', + ); + + return action; + } + + /** + * Complete a rebalance action by matching origin/destination/amount. + * Used by simulation when bridge delivers, since we don't have direct action ID correlation. + * Finds the oldest in-progress action that matches and completes it. + */ + completeRebalanceByRoute( + origin: Domain, + destination: Domain, + amount: bigint, + ): boolean { + // Find matching in-progress action (oldest first) + const allActions = Array.from(this.actions.values()); + const matchingActions = allActions + .filter( + (a) => + a.origin === origin && + a.destination === destination && + a.amount === amount && + a.status === 'in_progress', + ) + .sort((a, b) => a.createdAt - b.createdAt); + + if (matchingActions.length === 0) { + logger.debug( + { origin, destination, amount: amount.toString() }, + 'No matching in-progress action found for delivery', + ); + return false; + } + + const action = matchingActions[0]; + action.status = 'complete'; + action.updatedAt = Date.now(); + + // Update parent intent's fulfilledAmount + const intent = this.intents.get(action.intentId); + if (intent) { + intent.fulfilledAmount += action.amount; + intent.updatedAt = Date.now(); + + // If fully fulfilled, mark intent as complete + if (intent.fulfilledAmount >= intent.amount) { + intent.status = 'complete'; + logger.debug( + { + intentId: intent.id, + fulfilledAmount: intent.fulfilledAmount.toString(), + }, + 'Intent fully fulfilled, marking complete', + ); + } + } + + logger.debug( + { actionId: action.id, intentId: action.intentId, origin, destination }, + 'Completed rebalance action by route', + ); + return true; + } +} diff --git a/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts b/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts deleted file mode 100644 index e09d129f858..00000000000 --- a/typescript/rebalancer-sim/src/runners/MockInflightContextAdapter.ts +++ /dev/null @@ -1,141 +0,0 @@ -import type { - IInflightContextAdapter, - InflightContext, - Route, -} from '@hyperlane-xyz/rebalancer'; -import { rootLogger } from '@hyperlane-xyz/utils'; - -const logger = rootLogger.child({ module: 'MockInflightContextAdapter' }); - -/** - * Mock implementation of IInflightContextAdapter for simulation testing. - * - * This adapter maintains lists of pending rebalances and user transfers, - * allowing tests to control inflight context without needing a real - * ActionTracker or ExplorerClient. - * - * Usage: - * - Call addPendingRebalance() when a rebalance is initiated - * - Call addPendingTransfer() when a user transfer is initiated - * - Call removePendingRebalance() when a rebalance completes - * - Call removePendingTransfer() when a user transfer completes - * - The RebalancerService calls getInflightContext() during each poll cycle - */ -export class MockInflightContextAdapter implements IInflightContextAdapter { - private pendingRebalances: Array<{ id: string; route: Route }> = []; - private pendingTransfers: Array<{ id: string; route: Route }> = []; - - /** - * Add a pending rebalance (called when bridge transfer is initiated) - */ - addPendingRebalance(id: string, route: Route): void { - this.pendingRebalances.push({ id, route }); - logger.debug( - { - id, - origin: route.origin, - destination: route.destination, - amount: route.amount.toString(), - totalPending: this.pendingRebalances.length, - }, - 'Added pending rebalance', - ); - } - - /** - * Add a pending user transfer (called when user initiates warp transfer) - */ - addPendingTransfer(id: string, route: Route): void { - this.pendingTransfers.push({ id, route }); - logger.debug( - { - id, - origin: route.origin, - destination: route.destination, - amount: route.amount.toString(), - totalPending: this.pendingTransfers.length, - }, - 'Added pending transfer', - ); - } - - /** - * Remove a pending rebalance (called when bridge transfer completes) - */ - removePendingRebalance(id: string): boolean { - const idx = this.pendingRebalances.findIndex((r) => r.id === id); - if (idx >= 0) { - const removed = this.pendingRebalances.splice(idx, 1)[0]; - logger.debug( - { - id, - origin: removed.route.origin, - destination: removed.route.destination, - amount: removed.route.amount.toString(), - remainingPending: this.pendingRebalances.length, - }, - 'Removed pending rebalance', - ); - return true; - } - logger.warn({ id }, 'Attempted to remove non-existent pending rebalance'); - return false; - } - - /** - * Remove a pending user transfer (called when warp transfer delivers) - */ - removePendingTransfer(id: string): boolean { - const idx = this.pendingTransfers.findIndex((r) => r.id === id); - if (idx >= 0) { - const removed = this.pendingTransfers.splice(idx, 1)[0]; - logger.debug( - { - id, - origin: removed.route.origin, - destination: removed.route.destination, - amount: removed.route.amount.toString(), - remainingPending: this.pendingTransfers.length, - }, - 'Removed pending transfer', - ); - return true; - } - logger.warn({ id }, 'Attempted to remove non-existent pending transfer'); - return false; - } - - /** - * Get current inflight context for strategy decision-making. - * Called by RebalancerService during each poll cycle. - */ - async getInflightContext(): Promise { - return { - pendingRebalances: this.pendingRebalances.map((r) => r.route), - pendingTransfers: this.pendingTransfers.map((r) => r.route), - }; - } - - /** - * Clear all pending items (useful for test cleanup) - */ - clear(): void { - this.pendingRebalances = []; - this.pendingTransfers = []; - logger.debug('Cleared all pending items'); - } - - /** - * Get count of pending rebalances - */ - getPendingRebalanceCount(): number { - return this.pendingRebalances.length; - } - - /** - * Get count of pending transfers - */ - getPendingTransferCount(): number { - return this.pendingTransfers.length; - } -} diff --git a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts index 102589a0471..b33c73e3387 100644 --- a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts +++ b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts @@ -17,7 +17,7 @@ import type { RebalancerSimConfig, } from '../types.js'; -import { MockInflightContextAdapter } from './MockInflightContextAdapter.js'; +import { MockActionTracker } from './MockActionTracker.js'; import { SimulationRegistry } from './SimulationRegistry.js'; // Silent logger for the rebalancer service (internal) @@ -108,6 +108,15 @@ function buildStrategyConfig(config: RebalancerSimConfig): StrategyConfig { * ProductionRebalancerRunner runs the actual RebalancerService in-process. * This wraps the real CLI rebalancer for simulation testing. */ +/** + * Tracks pending rebalance info for correlating bridge callbacks with tracker actions. + */ +interface PendingRebalanceInfo { + origin: number; // domain ID + destination: number; // domain ID + amount: bigint; +} + export class ProductionRebalancerRunner extends EventEmitter implements IRebalancerRunner @@ -117,7 +126,9 @@ export class ProductionRebalancerRunner private config?: RebalancerSimConfig; private service?: RebalancerService; private running = false; - private mockAdapter?: MockInflightContextAdapter; + private mockTracker?: MockActionTracker; + // Maps bridge transfer ID to rebalance info for delivery correlation + private pendingRebalances = new Map(); async initialize(config: RebalancerSimConfig): Promise { // Cleanup any previously running instance @@ -125,8 +136,8 @@ export class ProductionRebalancerRunner this.config = config; - // Create mock inflight context adapter for simulation - this.mockAdapter = new MockInflightContextAdapter(); + // Create mock action tracker for simulation + this.mockTracker = new MockActionTracker(); } async start(): Promise { @@ -226,7 +237,7 @@ export class ProductionRebalancerRunner strategyConfig, ] as StrategyConfig[]); - // Create service with mock inflight context adapter + // Create service with mock action tracker this.service = new RebalancerService( multiProvider, multiProtocolProvider, @@ -238,7 +249,7 @@ export class ProductionRebalancerRunner monitorOnly: false, withMetrics: false, logger: silentLogger, - inflightContextAdapter: this.mockAdapter, + actionTracker: this.mockTracker, }, ); @@ -284,10 +295,11 @@ export class ProductionRebalancerRunner } this.config = undefined; - if (this.mockAdapter) { - this.mockAdapter.clear(); - this.mockAdapter = undefined; + if (this.mockTracker) { + this.mockTracker.clear(); + this.mockTracker = undefined; } + this.pendingRebalances.clear(); this.removeAllListeners(); } @@ -306,10 +318,19 @@ export class ProductionRebalancerRunner * SimulationEngine uses these to notify about pending transfers and rebalances. */ getInflightCallbacks(): InflightContextCallbacks | undefined { - if (!this.mockAdapter) { + if (!this.mockTracker || !this.config) { return undefined; } + // Helper to convert chain name to domain ID + const getDomainId = (chainName: string): number => { + const domain = this.config!.deployment.domains[chainName]; + if (!domain) { + throw new Error(`Unknown chain: ${chainName}`); + } + return domain.domainId; + }; + return { onTransferInitiated: ( id: string, @@ -317,14 +338,15 @@ export class ProductionRebalancerRunner destination: string, amount: bigint, ) => { - this.mockAdapter!.addPendingTransfer(id, { - origin, - destination, + this.mockTracker!.addTransfer( + id, + getDomainId(origin), + getDomainId(destination), amount, - }); + ); }, onTransferDelivered: (id: string) => { - this.mockAdapter!.removePendingTransfer(id); + this.mockTracker!.removeTransfer(id); }, onRebalanceInitiated: ( id: string, @@ -332,14 +354,37 @@ export class ProductionRebalancerRunner destination: string, amount: bigint, ) => { - this.mockAdapter!.addPendingRebalance(id, { - origin, - destination, + const originDomain = getDomainId(origin); + const destDomain = getDomainId(destination); + + // Store mapping from bridge transfer ID to route info for delivery correlation + this.pendingRebalances.set(id, { + origin: originDomain, + destination: destDomain, amount, }); + + // Create action for the pending intent. + // RebalancerService can't extract messageId from MockValueTransferBridge + // (no Dispatch event), so we create the action ourselves. + this.mockTracker!.createActionForPendingIntent( + originDomain, + destDomain, + amount, + id, + ); }, onRebalanceDelivered: (id: string) => { - this.mockAdapter!.removePendingRebalance(id); + // Look up the route info and complete the matching action in tracker + const info = this.pendingRebalances.get(id); + if (info) { + this.pendingRebalances.delete(id); + this.mockTracker!.completeRebalanceByRoute( + info.origin, + info.destination, + info.amount, + ); + } }, }; } diff --git a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts index bfc24e6580d..2640b5200e0 100644 --- a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts +++ b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts @@ -29,8 +29,7 @@ import { } from '../utils/simulation-helpers.js'; describe('Rebalancer Simulation', function () { - const anvilPort = 8545; - const anvil = setupAnvilTestSuite(this, anvilPort); + const anvil = setupAnvilTestSuite(this); before(async function () { ensureResultsDir(); diff --git a/typescript/rebalancer-sim/test/integration/harness-setup.test.ts b/typescript/rebalancer-sim/test/integration/harness-setup.test.ts index 66edf883f5d..9a395c07be9 100644 --- a/typescript/rebalancer-sim/test/integration/harness-setup.test.ts +++ b/typescript/rebalancer-sim/test/integration/harness-setup.test.ts @@ -37,11 +37,11 @@ import { import { setupAnvilTestSuite } from '../utils/anvil.js'; describe('Multi-Domain Deployment', function () { - const anvilPort = 8546; // Use different port to avoid conflict with other tests - const anvil = setupAnvilTestSuite(this, anvilPort); + const anvil = setupAnvilTestSuite(this); let provider: ethers.providers.JsonRpcProvider; - before(async () => { + beforeEach(async () => { + // Create provider after anvil starts (rpc is set in beforeEach) provider = new ethers.providers.JsonRpcProvider(anvil.rpc); }); diff --git a/typescript/rebalancer-sim/test/utils/anvil.ts b/typescript/rebalancer-sim/test/utils/anvil.ts index 5495086f7ec..c821c10f3ab 100644 --- a/typescript/rebalancer-sim/test/utils/anvil.ts +++ b/typescript/rebalancer-sim/test/utils/anvil.ts @@ -6,15 +6,16 @@ import { import { retryAsync } from '@hyperlane-xyz/utils'; -const DEFAULT_ANVIL_PORT = 8545; +const CONTAINER_PORT = 8545; // Port inside the container (fixed) const DEFAULT_CHAIN_ID = 31337; /** * Start an Anvil container using testcontainers. - * Uses the same pattern as CLI e2e tests for consistency. + * Uses dynamic port assignment - testcontainers picks an available host port. + * + * @returns The started container. Use container.getMappedPort(8545) to get the host port. */ export async function startAnvilContainer( - port: number = DEFAULT_ANVIL_PORT, chainId: number = DEFAULT_CHAIN_ID, ): Promise { return retryAsync( @@ -25,14 +26,11 @@ export async function startAnvilContainer( '--host', '0.0.0.0', '-p', - port.toString(), + CONTAINER_PORT.toString(), '--chain-id', chainId.toString(), ]) - .withExposedPorts({ - container: port, - host: port, - }) + .withExposedPorts(CONTAINER_PORT) // Dynamic host port assignment .withWaitStrategy(Wait.forLogMessage(/Listening on/)) .start(), 3, // maxRetries @@ -40,6 +38,15 @@ export async function startAnvilContainer( ); } +/** + * Get the RPC URL for a started anvil container. + */ +export function getAnvilRpcUrl(container: StartedTestContainer): string { + const host = container.getHost(); + const port = container.getMappedPort(CONTAINER_PORT); + return `http://${host}:${port}`; +} + /** * Setup function for Mocha tests that require Anvil. * Starts a fresh Anvil container for EACH TEST to ensure complete isolation. @@ -47,28 +54,32 @@ export async function startAnvilContainer( * Uses testcontainers for: * - No local anvil installation required * - Automatic container cleanup (even on crashes) + * - Dynamic port assignment (no port conflicts) * - Retry logic for CI reliability * - Consistent behavior across local/CI environments * * Usage: * ```typescript * describe('My Tests', function() { - * const anvil = setupAnvilTestSuite(this, 8545); + * const anvil = setupAnvilTestSuite(this); * * it('test case', async () => { - * const rpc = anvil.rpc; // http://127.0.0.1:8545 + * const rpc = anvil.rpc; // http://localhost: * }); * }); * ``` */ export function setupAnvilTestSuite( suite: Mocha.Suite, - port: number = DEFAULT_ANVIL_PORT, chainId: number = DEFAULT_CHAIN_ID, ): { rpc: string } { - const state: { rpc: string; container: StartedTestContainer | null } = { - rpc: `http://127.0.0.1:${port}`, + // Use a getter pattern so rpc is always current after container starts + const state: { + container: StartedTestContainer | null; + rpc: string; + } = { container: null, + rpc: '', // Will be set after container starts }; suite.timeout(180000); // 3 minutes per test @@ -81,7 +92,8 @@ export function setupAnvilTestSuite( state.container = null; } - state.container = await startAnvilContainer(port, chainId); + state.container = await startAnvilContainer(chainId); + state.rpc = getAnvilRpcUrl(state.container); }); // Stop container after EACH test for clean slate diff --git a/typescript/rebalancer/src/core/RebalancerService.ts b/typescript/rebalancer/src/core/RebalancerService.ts index d06a448ace2..05a73dd7951 100644 --- a/typescript/rebalancer/src/core/RebalancerService.ts +++ b/typescript/rebalancer/src/core/RebalancerService.ts @@ -36,7 +36,7 @@ import { Metrics } from '../metrics/Metrics.js'; import { Monitor } from '../monitor/Monitor.js'; import { type IActionTracker, - type IInflightContextAdapter, + InflightContextAdapter, } from '../tracking/index.js'; import { getRawBalances } from '../utils/balanceUtils.js'; @@ -63,11 +63,11 @@ export interface RebalancerServiceConfig { version?: string; /** - * Optional pre-configured inflight context adapter. - * If provided, skips ActionTracker creation and uses this adapter directly. - * Useful for simulation/testing where inflight context is mocked externally. + * Optional pre-configured ActionTracker. + * If provided, skips ActionTracker creation and uses this directly. + * Useful for simulation/testing where tracking is mocked externally. */ - inflightContextAdapter?: IInflightContextAdapter; + actionTracker?: IActionTracker; } export interface ManualRebalanceRequest { @@ -127,7 +127,7 @@ export class RebalancerService { private metrics?: Metrics; private mode: 'manual' | 'daemon'; private actionTracker?: IActionTracker; - private inflightContextAdapter?: IInflightContextAdapter; + private inflightContextAdapter?: InflightContextAdapter; constructor( private readonly multiProvider: MultiProvider, @@ -186,15 +186,16 @@ export class RebalancerService { ); } - // Create or use provided inflight context adapter - if (this.config.inflightContextAdapter) { - // Use externally provided adapter (e.g., for simulation/testing) - this.inflightContextAdapter = this.config.inflightContextAdapter; - this.logger.info( - 'Using externally provided InflightContextAdapter (no ActionTracker)', + // Create or use provided ActionTracker for tracking inflight actions + if (this.config.actionTracker) { + // Use externally provided ActionTracker (e.g., for simulation/testing) + this.actionTracker = this.config.actionTracker; + this.inflightContextAdapter = new InflightContextAdapter( + this.actionTracker, + this.multiProvider, ); + this.logger.info('Using externally provided ActionTracker'); } else { - // Create ActionTracker for tracking inflight actions const { tracker, adapter } = await this.contextFactory.createActionTracker(); this.actionTracker = tracker; @@ -434,21 +435,12 @@ export class RebalancerService { /** * Execute rebalancing with intent tracking. * Creates intents and assigns IDs to routes before execution, then processes results by ID. - * If ActionTracker is not available (e.g., using external inflightContextAdapter), - * falls back to simple execution without tracking. */ private async executeWithTracking( strategyRoutes: StrategyRoute[], ): Promise { - if (!this.rebalancer) { - this.logger.warn('Rebalancer not available, skipping'); - return; - } - - // Fall back to simple execution if ActionTracker is not available - // (e.g., when using an external inflightContextAdapter for simulation) - if (!this.actionTracker) { - await this.executeWithoutTracking(strategyRoutes); + if (!this.rebalancer || !this.actionTracker) { + this.logger.warn('Rebalancer or ActionTracker not available, skipping'); return; } @@ -553,73 +545,6 @@ export class RebalancerService { } } - /** - * Execute rebalancing without ActionTracker tracking. - * Used when an external inflightContextAdapter is provided (e.g., simulation). - * Simply executes the routes and logs results. - */ - private async executeWithoutTracking( - strategyRoutes: StrategyRoute[], - ): Promise { - assert(this.rebalancer, 'Rebalancer must be available'); - - // Generate simple IDs for routes (not tracked in ActionTracker) - const rebalanceRoutes: RebalanceRoute[] = strategyRoutes.map((route) => ({ - ...route, - intentId: randomUUID(), - })); - - this.logger.debug( - { routeCount: rebalanceRoutes.length }, - 'Executing rebalance routes (without tracking)', - ); - - try { - const results = await this.rebalancer.rebalance(rebalanceRoutes); - const failedResults = results.filter((r) => !r.success); - if (failedResults.length > 0) { - this.metrics?.recordRebalancerFailure(); - this.logger.warn( - { failureCount: failedResults.length, total: results.length }, - 'Rebalancer cycle completed with failures', - ); - } else { - this.metrics?.recordRebalancerSuccess(); - this.logger.info( - { routeCount: results.length }, - 'Rebalancer completed a cycle successfully (no tracking)', - ); - } - - // Log results for debugging - for (const result of results) { - if (result.success) { - this.logger.debug( - { - origin: result.route.origin, - destination: result.route.destination, - messageId: result.messageId, - txHash: result.txHash, - }, - 'Rebalance executed successfully', - ); - } else { - this.logger.warn( - { - origin: result.route.origin, - destination: result.route.destination, - error: result.error, - }, - 'Rebalance execution failed', - ); - } - } - } catch (error: any) { - this.metrics?.recordRebalancerFailure(); - this.logger.error({ error }, 'Error while rebalancing'); - } - } - /** * Event handler for monitor errors */ diff --git a/typescript/rebalancer/src/index.ts b/typescript/rebalancer/src/index.ts index f65300b6b65..1c6bfaa95b8 100644 --- a/typescript/rebalancer/src/index.ts +++ b/typescript/rebalancer/src/index.ts @@ -65,7 +65,6 @@ export type { export type { IStrategy, StrategyRoute, - Route, RawBalances, InflightContext, } from './interfaces/IStrategy.js'; @@ -93,10 +92,12 @@ export { isCollateralizedTokenEligibleForRebalancing } from './utils/tokenUtils. export { ExplorerClient } from './utils/ExplorerClient.js'; // Tracking -export { - type IInflightContextAdapter, - InflightContextAdapter, -} from './tracking/InflightContextAdapter.js'; +export { InflightContextAdapter } from './tracking/InflightContextAdapter.js'; +export type { + IActionTracker, + CreateRebalanceIntentParams, + CreateRebalanceActionParams, +} from './tracking/IActionTracker.js'; // Factory export { RebalancerContextFactory } from './factories/RebalancerContextFactory.js'; diff --git a/typescript/rebalancer/src/tracking/InflightContextAdapter.ts b/typescript/rebalancer/src/tracking/InflightContextAdapter.ts index 6874d1ec69a..fac6be9228c 100644 --- a/typescript/rebalancer/src/tracking/InflightContextAdapter.ts +++ b/typescript/rebalancer/src/tracking/InflightContextAdapter.ts @@ -4,20 +4,11 @@ import type { InflightContext } from '../interfaces/IStrategy.js'; import type { IActionTracker } from './IActionTracker.js'; -/** - * Interface for adapters that provide inflight context to strategies. - * This allows mocking inflight context in simulations without needing - * a full ActionTracker/ExplorerClient setup. - */ -export interface IInflightContextAdapter { - getInflightContext(): Promise; -} - /** * Adapter that converts ActionTracker data to strategy-consumable InflightContext. * Handles conversion from Domain IDs (used by ActionTracker) to ChainNames (used by Strategy). */ -export class InflightContextAdapter implements IInflightContextAdapter { +export class InflightContextAdapter { constructor( private readonly actionTracker: IActionTracker, private readonly multiProvider: MultiProvider, diff --git a/typescript/rebalancer/src/tracking/index.ts b/typescript/rebalancer/src/tracking/index.ts index 3a5b1507c54..46a295c168e 100644 --- a/typescript/rebalancer/src/tracking/index.ts +++ b/typescript/rebalancer/src/tracking/index.ts @@ -33,7 +33,4 @@ export type { } from './IActionTracker.js'; // Export InflightContextAdapter -export { - type IInflightContextAdapter, - InflightContextAdapter, -} from './InflightContextAdapter.js'; +export { InflightContextAdapter } from './InflightContextAdapter.js'; From d150d7414752a54a43505a8593126cdf1332e001 Mon Sep 17 00:00:00 2001 From: nambrot Date: Wed, 4 Feb 2026 15:52:28 -0500 Subject: [PATCH 04/12] test(rebalancer-sim): Add assertions for inflight tracking effectiveness - inflight-guard: Assert ProductionRebalancer uses >50% fewer rebalances than SimpleRebalancer (currently achieves ~75% reduction) - Update test description to reflect working inflight tracking - Update blocked-user-transfer comment to use correct terminology Co-Authored-By: Claude Opus 4.5 --- .../test/integration/full-simulation.test.ts | 55 +++++++++++++++---- 1 file changed, 45 insertions(+), 10 deletions(-) diff --git a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts index 2640b5200e0..c8d5cebceb6 100644 --- a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts +++ b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts @@ -222,12 +222,11 @@ describe('Rebalancer Simulation', function () { * and polling at 1000ms, a rebalancer without inflight awareness will * over-rebalance because it doesn't account for pending transfers. * - * CURRENT LIMITATION: ProductionRebalancerService tracks self-created rebalance - * intents via ActionTracker's in-memory store, but user transfer tracking relies - * on ExplorerClient which has no indexed data in simulation (Anvil). - * Until a mock ExplorerClient is implemented, this test runs as report-only. + * SimpleRebalancer: No inflight tracking, over-rebalances significantly + * ProductionRebalancer: Tracks pending rebalances via MockActionTracker, + * significantly reduces redundant rebalances (typically 60-80% fewer) */ - it('inflight-guard: demonstrates over-rebalancing with slow bridge (report only)', async function () { + it('inflight-guard: ProductionRebalancer uses fewer rebalances with inflight tracking', async function () { // This test takes longer due to 3s bridge delays and runs multiple rebalancers this.timeout(120000); @@ -235,17 +234,53 @@ describe('Rebalancer Simulation', function () { anvilRpc: anvil.rpc, }); - // Report results - demonstrates over-rebalancing behavior + // Report results console.log('\n INFLIGHT GUARD REPORT:'); for (const result of results) { console.log( ` ${result.rebalancerName}: ${result.kpis.totalRebalances} rebalances`, ); } - console.log(' (Expected with proper inflight tracking: ≤2 rebalances)'); - console.log( - ' Note: Currently both over-rebalance due to Explorer unavailability in simulation', + + // Find results by rebalancer type + const productionResult = results.find( + (r) => r.rebalancerName === 'ProductionRebalancerService', ); + const simpleResult = results.find( + (r) => r.rebalancerName === 'SimpleRebalancer', + ); + + // Both should complete all transfers + if (productionResult) { + expect(productionResult.kpis.completionRate).to.equal( + 1, + 'ProductionRebalancer should complete all transfers', + ); + } + if (simpleResult) { + expect(simpleResult.kpis.completionRate).to.equal( + 1, + 'SimpleRebalancer should complete all transfers', + ); + } + + // ProductionRebalancer with inflight tracking should use significantly fewer rebalances + if (productionResult && simpleResult) { + expect(productionResult.kpis.totalRebalances).to.be.lessThan( + simpleResult.kpis.totalRebalances, + 'ProductionRebalancer should use fewer rebalances than SimpleRebalancer', + ); + + // ProductionRebalancer should use at most 50% of SimpleRebalancer's rebalances + // (typically achieves 60-80% reduction) + const reductionRatio = + productionResult.kpis.totalRebalances / + simpleResult.kpis.totalRebalances; + expect(reductionRatio).to.be.lessThan( + 0.5, + `ProductionRebalancer should achieve >50% reduction in rebalances (got ${((1 - reductionRatio) * 100).toFixed(0)}% reduction)`, + ); + } }); // BLOCKED USER TRANSFER @@ -264,7 +299,7 @@ describe('Rebalancer Simulation', function () { * SimpleRebalancer: Only sees on-chain balances, doesn't know about pending * transfer, weights appear within tolerance → no action → transfer stuck * - * ProductionRebalancer: Mock adapter tracks pending transfer, strategy + * ProductionRebalancer: MockActionTracker tracks pending transfer, strategy * reserves collateral for it, detects deficit → rebalances → transfer succeeds */ it('blocked-user-transfer: ProductionRebalancer proactively adds collateral for pending transfers', async function () { From 6fa2db9cac8a0dfb4b91d9f96881c4f227cc77d3 Mon Sep 17 00:00:00 2001 From: Nam Chu Hoai Date: Fri, 6 Feb 2026 09:51:33 -0500 Subject: [PATCH 05/12] refactor(rebalancer-sim): Unify mocking in MockInfrastructureController (#8060) Co-authored-by: Claude Opus 4.6 --- .../mock/MockValueTransferBridge.sol | 41 +- .../ethereum/warp/warp-rebalancer.e2e-test.ts | 8 +- .../scenarios/inflight-guard.json | 24 +- .../src/BridgeMockController.ts | 404 ------------------ .../rebalancer-sim/src/MessageTracker.ts | 312 -------------- .../src/MockInfrastructureController.ts | 346 +++++++++++++++ .../src/RebalancerSimulationHarness.ts | 9 - .../src/SimulationDeployment.ts | 29 +- .../rebalancer-sim/src/SimulationEngine.ts | 301 ++++--------- typescript/rebalancer-sim/src/index.ts | 3 +- .../src/runners/MockActionTracker.ts | 14 +- .../src/runners/ProductionRebalancerRunner.ts | 104 +---- typescript/rebalancer-sim/src/types.ts | 47 +- .../test/integration/full-simulation.test.ts | 28 -- 14 files changed, 511 insertions(+), 1159 deletions(-) delete mode 100644 typescript/rebalancer-sim/src/BridgeMockController.ts delete mode 100644 typescript/rebalancer-sim/src/MessageTracker.ts create mode 100644 typescript/rebalancer-sim/src/MockInfrastructureController.ts diff --git a/solidity/contracts/mock/MockValueTransferBridge.sol b/solidity/contracts/mock/MockValueTransferBridge.sol index cdc779a1a5e..bc69d292295 100644 --- a/solidity/contracts/mock/MockValueTransferBridge.sol +++ b/solidity/contracts/mock/MockValueTransferBridge.sol @@ -2,17 +2,15 @@ pragma solidity ^0.8.13; import {ITokenBridge, Quote} from "../interfaces/ITokenBridge.sol"; +import {Router} from "../client/Router.sol"; import {IERC20} from "@openzeppelin/contracts/token/ERC20/IERC20.sol"; import {SafeERC20} from "@openzeppelin/contracts/token/ERC20/utils/SafeERC20.sol"; +import {ERC20Test} from "../test/ERC20Test.sol"; -contract MockValueTransferBridge is ITokenBridge { +contract MockValueTransferBridge is Router, ITokenBridge { using SafeERC20 for IERC20; address public collateral; - constructor(address _collateral) { - collateral = _collateral; - } - event SentTransferRemote( uint32 indexed origin, uint32 indexed destination, @@ -20,6 +18,18 @@ contract MockValueTransferBridge is ITokenBridge { uint256 amount ); + constructor(address _collateral, address _mailbox) Router(_mailbox) { + collateral = _collateral; + } + + function initialize( + address _hook, + address _ism, + address _owner + ) external initializer { + _MailboxClient_initialize(_hook, _ism, _owner); + } + function quoteTransferRemote( uint32, //_destinationDomain, bytes32, //_recipient, @@ -50,6 +60,25 @@ contract MockValueTransferBridge is ITokenBridge { _amountOut ); - return keccak256("transferId"); + // Dispatch through MockMailbox so Dispatch event is emitted + transferId = _Router_dispatch( + _destinationDomain, + msg.value, + abi.encode(_recipient, _amountOut) + ); + } + + function _handle( + uint32, // _origin + bytes32, // _sender + bytes calldata _message + ) internal virtual override { + (bytes32 recipientBytes32, uint256 amount) = abi.decode( + _message, + (bytes32, uint256) + ); + address recipient = address(uint160(uint256(recipientBytes32))); + // Mint collateral tokens to recipient (destination warp token) + ERC20Test(collateral).mintTo(recipient, amount); } } diff --git a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts index 7f4fe79e65e..5b4dcf7c0a7 100644 --- a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts +++ b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts @@ -823,7 +823,7 @@ describe('hyperlane warp rebalancer e2e tests', async function () { // Deploy the bridge const bridgeContract = await new MockValueTransferBridge__factory( chain3Signer, - ).deploy(tokenChain3.address); + ).deploy(tokenChain3.address, chain3Addresses.mailbox); // Allow bridge await chain3CollateralContract.addBridge( @@ -877,7 +877,7 @@ describe('hyperlane warp rebalancer e2e tests', async function () { // Deploy the bridge const bridgeContract = await new MockValueTransferBridge__factory( chain3Signer, - ).deploy(tokenChain3.address); + ).deploy(tokenChain3.address, chain3Addresses.mailbox); // Allow bridge await chain3CollateralContract.addBridge( @@ -961,7 +961,7 @@ describe('hyperlane warp rebalancer e2e tests', async function () { // It will also allow us to mock some token movement const bridgeContract = await new MockValueTransferBridge__factory( originSigner, - ).deploy(tokenChain3.address); + ).deploy(tokenChain3.address, chain3Addresses.mailbox); // Allow bridge // This allow the bridge to be used to send the rebalance transaction @@ -1208,7 +1208,7 @@ describe('hyperlane warp rebalancer e2e tests', async function () { // It will also allow us to mock some token movement const bridgeContract = await new MockValueTransferBridge__factory( originSigner, - ).deploy(tokenChain3.address); + ).deploy(tokenChain3.address, chain3Addresses.mailbox); // Allow bridge // This allow the bridge to be used to send the rebalance transaction diff --git a/typescript/rebalancer-sim/scenarios/inflight-guard.json b/typescript/rebalancer-sim/scenarios/inflight-guard.json index 8d106313a17..bcbcc030c79 100644 --- a/typescript/rebalancer-sim/scenarios/inflight-guard.json +++ b/typescript/rebalancer-sim/scenarios/inflight-guard.json @@ -1,8 +1,8 @@ { "name": "inflight-guard", "description": "Tests rebalancer behavior with slow bridge and fast polling to demonstrate inflight guard importance.", - "expectedBehavior": "With SLOW bridge (3s) and polling at 1000ms, a rebalancer without inflight tracking will over-rebalance.\nWithout inflight guard: Multiple redundant transfers sent before first one delivers.\nWith inflight guard: Only 1-2 transfers sent, tracking pending amounts.", - "duration": 8000, + "expectedBehavior": "With SLOW bridge (4s) and polling at 1000ms, a rebalancer without inflight tracking will over-rebalance.\nWithout inflight guard: Multiple redundant transfers sent before first one delivers.\nWith inflight guard: Only 1-2 transfers sent, tracking pending amounts.", + "duration": 10000, "chains": ["chain1", "chain2"], "initialImbalance": { "chain1": "50000000000000000000" @@ -18,27 +18,11 @@ }, { "id": "keepalive-2", - "timestamp": 3000, - "origin": "chain1", - "destination": "chain2", - "amount": "40000000000000000000", - "user": "0x1111111111111111111111111111111111111111" - }, - { - "id": "keepalive-3", "timestamp": 5000, "origin": "chain1", "destination": "chain2", "amount": "40000000000000000000", "user": "0x1111111111111111111111111111111111111111" - }, - { - "id": "keepalive-4", - "timestamp": 7000, - "origin": "chain1", - "destination": "chain2", - "amount": "40000000000000000000", - "user": "0x1111111111111111111111111111111111111111" } ], "defaultInitialCollateral": "100000000000000000000", @@ -50,14 +34,14 @@ "defaultBridgeConfig": { "chain1": { "chain2": { - "deliveryDelay": 3000, + "deliveryDelay": 4000, "failureRate": 0, "deliveryJitter": 0 } }, "chain2": { "chain1": { - "deliveryDelay": 3000, + "deliveryDelay": 4000, "failureRate": 0, "deliveryJitter": 0 } diff --git a/typescript/rebalancer-sim/src/BridgeMockController.ts b/typescript/rebalancer-sim/src/BridgeMockController.ts deleted file mode 100644 index 149794c3062..00000000000 --- a/typescript/rebalancer-sim/src/BridgeMockController.ts +++ /dev/null @@ -1,404 +0,0 @@ -import { ethers } from 'ethers'; -import { EventEmitter } from 'events'; - -import { - ERC20Test__factory, - MockValueTransferBridge__factory, -} from '@hyperlane-xyz/core'; -import type { Address } from '@hyperlane-xyz/utils'; -import { rootLogger } from '@hyperlane-xyz/utils'; - -import type { - BridgeEvent, - BridgeMockConfig, - BridgeRouteConfig, - DeployedDomain, - PendingTransfer, -} from './types.js'; -import { DEFAULT_BRIDGE_ROUTE_CONFIG } from './types.js'; - -const logger = rootLogger.child({ module: 'BridgeMockController' }); - -/** - * BridgeMockController manages simulated bridge transfers with configurable - * delays, failures, and fees. It intercepts SentTransferRemote events and - * schedules async delivery to simulate real bridge behavior. - */ -export class BridgeMockController extends EventEmitter { - private pendingTransfers: Map = new Map(); - private completedTransfers: PendingTransfer[] = []; - private transferCounter = 0; - private deliveryTimers: Map = new Map(); - private isRunning = false; - private eventListeners: Map = new Map(); - - // Transaction queue to prevent nonce collisions - private txQueue: Array<() => Promise> = []; - private txProcessing = false; - - constructor( - private readonly provider: ethers.providers.JsonRpcProvider, - private readonly domains: Record, - private readonly deployerKey: string, - private readonly bridgeConfig: BridgeMockConfig = {}, - ) { - super(); - } - - /** - * Queue a transaction to be executed serially (prevents nonce collisions) - */ - private async queueTransaction(fn: () => Promise): Promise { - return new Promise((resolve, reject) => { - this.txQueue.push(async () => { - try { - await fn(); - resolve(); - } catch (error) { - reject(error); - } - }); - void this.processQueue(); - }); - } - - /** - * Process queued transactions one at a time - */ - private async processQueue(): Promise { - if (this.txProcessing || this.txQueue.length === 0) return; - - this.txProcessing = true; - while (this.txQueue.length > 0) { - const fn = this.txQueue.shift(); - if (fn) { - try { - await fn(); - } catch (_error) { - // Error already handled in queueTransaction - } - } - } - this.txProcessing = false; - } - - /** - * Gets the bridge config for a specific route - */ - private getRouteConfig( - origin: string, - destination: string, - ): BridgeRouteConfig { - return ( - this.bridgeConfig[origin]?.[destination] ?? DEFAULT_BRIDGE_ROUTE_CONFIG - ); - } - - /** - * Calculates delivery delay with jitter - */ - private calculateDelay(config: BridgeRouteConfig): number { - const jitter = (Math.random() - 0.5) * 2 * config.deliveryJitter; - return Math.max(0, config.deliveryDelay + jitter); - } - - /** - * Start listening for bridge events - */ - async start(): Promise { - if (this.isRunning) return; - this.isRunning = true; - - const deployer = new ethers.Wallet(this.deployerKey, this.provider); - - // Set up event listeners for each bridge - for (const [chainName, domain] of Object.entries(this.domains)) { - const bridge = MockValueTransferBridge__factory.connect( - domain.bridge, - deployer, - ); - - // Listen for SentTransferRemote events - bridge.on( - bridge.filters.SentTransferRemote(), - (origin, destination, recipient, amount) => { - void this.onTransferInitiated( - chainName, - origin, - destination, - recipient, - amount.toBigInt(), - ); - }, - ); - - this.eventListeners.set(chainName, bridge); - } - } - - /** - * Stop listening and cancel pending deliveries - */ - async stop(): Promise { - this.isRunning = false; - - // Remove event listeners - for (const bridge of this.eventListeners.values()) { - bridge.removeAllListeners(); - } - this.eventListeners.clear(); - - // Cancel pending delivery timers - for (const timer of this.deliveryTimers.values()) { - clearTimeout(timer); - } - this.deliveryTimers.clear(); - } - - /** - * Handle transfer initiated event - */ - private async onTransferInitiated( - originChain: string, - originDomainId: number, - destinationDomainId: number, - recipientBytes32: string, - amount: bigint, - ): Promise { - // Find destination chain by domain ID - const destChain = Object.entries(this.domains).find( - ([_, d]) => d.domainId === destinationDomainId, - )?.[0]; - - if (!destChain) { - logger.error({ destinationDomainId }, 'Unknown destination domain'); - return; - } - - const config = this.getRouteConfig(originChain, destChain); - const transferId = `bridge-${this.transferCounter++}`; - const delay = this.calculateDelay(config); - - // Apply token fee if configured - let netAmount = amount; - if (config.tokenFeeBps) { - const fee = (amount * BigInt(config.tokenFeeBps)) / BigInt(10000); - netAmount = amount - fee; - } - - const recipient = ethers.utils.hexDataSlice( - recipientBytes32, - 12, - ) as Address; - - // MockValueTransferBridge pulls tokens from origin warp token. - // Bridge delivery mints to destination, preserving total warp token collateral. - - const pendingTransfer: PendingTransfer = { - id: transferId, - origin: originChain, - destination: destChain, - amount: netAmount, - recipient, - scheduledDelivery: Date.now() + delay, - failed: false, - delivered: false, - }; - - this.pendingTransfers.set(transferId, pendingTransfer); - - // Emit event - const bridgeEvent: BridgeEvent = { - type: 'transfer_initiated', - transfer: pendingTransfer, - timestamp: Date.now(), - }; - this.emit('transfer_initiated', bridgeEvent); - - // Schedule delivery - const timer = setTimeout( - () => this.executeDelivery(transferId, config), - delay, - ); - this.deliveryTimers.set(transferId, timer); - } - - /** - * Execute delivery of a pending transfer - */ - private async executeDelivery( - transferId: string, - config: BridgeRouteConfig, - ): Promise { - const transfer = this.pendingTransfers.get(transferId); - if (!transfer || transfer.delivered) return; - - this.deliveryTimers.delete(transferId); - - // Check for failure - if (Math.random() < config.failureRate) { - transfer.failed = true; - this.pendingTransfers.delete(transferId); - this.completedTransfers.push(transfer); - - const event: BridgeEvent = { - type: 'transfer_failed', - transfer, - timestamp: Date.now(), - }; - this.emit('transfer_failed', event); - return; - } - - try { - // Execute the delivery by simulating tokens arriving at destination - // In a real scenario, this would call the destination warp token's handle function - // For simulation, we directly transfer tokens to simulate bridge completion - await this.simulateBridgeDelivery(transfer); - - transfer.delivered = true; - transfer.deliveredAt = Date.now(); - this.pendingTransfers.delete(transferId); - this.completedTransfers.push(transfer); - - const event: BridgeEvent = { - type: 'transfer_delivered', - transfer, - timestamp: Date.now(), - }; - this.emit('transfer_delivered', event); - } catch (error) { - logger.error({ transferId, error }, 'Bridge delivery failed'); - transfer.failed = true; - this.pendingTransfers.delete(transferId); - this.completedTransfers.push(transfer); - - const event: BridgeEvent = { - type: 'transfer_failed', - transfer, - timestamp: Date.now(), - }; - this.emit('transfer_failed', event); - } - } - - /** - * Simulate bridge delivery by burning from origin bridge and minting to destination. - * This maintains token conservation across the simulation. - * Uses transaction queue to prevent nonce collisions. - */ - private async simulateBridgeDelivery( - transfer: PendingTransfer, - ): Promise { - await this.queueTransaction(async () => { - const deployer = new ethers.Wallet(this.deployerKey, this.provider); - const originDomain = this.domains[transfer.origin]; - const destDomain = this.domains[transfer.destination]; - - // Burn from origin bridge (tokens are guaranteed to be there since SentTransferRemote fired) - const originCollateralToken = ERC20Test__factory.connect( - originDomain.collateralToken, - deployer, - ); - const burnTx = await originCollateralToken.burnFrom( - originDomain.bridge, - transfer.amount.toString(), - ); - await burnTx.wait(); - - // Mint same amount to destination warp token - const destCollateralToken = ERC20Test__factory.connect( - destDomain.collateralToken, - deployer, - ); - const mintTx = await destCollateralToken.mintTo( - destDomain.warpToken, - transfer.amount.toString(), - ); - await mintTx.wait(); - }); - } - - /** - * Manually trigger delivery for a pending transfer (for testing) - */ - async forceDelivery(transferId: string): Promise { - const transfer = this.pendingTransfers.get(transferId); - if (!transfer) { - throw new Error(`Transfer not found: ${transferId}`); - } - - // Cancel scheduled delivery - const timer = this.deliveryTimers.get(transferId); - if (timer) { - clearTimeout(timer); - this.deliveryTimers.delete(transferId); - } - - // Execute immediately - await this.executeDelivery( - transferId, - this.getRouteConfig(transfer.origin, transfer.destination), - ); - } - - /** - * Check if there are pending transfers - */ - hasPendingTransfers(): boolean { - return this.pendingTransfers.size > 0; - } - - /** - * Get count of pending transfers - */ - getPendingCount(): number { - return this.pendingTransfers.size; - } - - /** - * Get all pending transfers - */ - getPendingTransfers(): PendingTransfer[] { - return Array.from(this.pendingTransfers.values()); - } - - /** - * Get completed transfers - */ - getCompletedTransfers(): PendingTransfer[] { - return [...this.completedTransfers]; - } - - /** - * Wait for all pending transfers to complete - * On timeout, marks remaining transfers as failed and clears them - */ - async waitForAllDeliveries(timeoutMs: number = 30000): Promise { - const startTime = Date.now(); - - while (this.hasPendingTransfers()) { - if (Date.now() - startTime > timeoutMs) { - const pendingCount = this.getPendingCount(); - logger.warn( - { pendingCount }, - 'Timeout waiting for bridge deliveries - marking as failed', - ); - // Mark all pending as failed, update state, and clear - for (const transfer of this.pendingTransfers.values()) { - transfer.failed = true; - this.completedTransfers.push(transfer); - const event: BridgeEvent = { - type: 'transfer_failed', - transfer, - timestamp: Date.now(), - }; - this.emit('transfer_failed', event); - } - this.pendingTransfers.clear(); - break; - } - await new Promise((resolve) => setTimeout(resolve, 100)); - } - } -} diff --git a/typescript/rebalancer-sim/src/MessageTracker.ts b/typescript/rebalancer-sim/src/MessageTracker.ts deleted file mode 100644 index 87ff3f066d5..00000000000 --- a/typescript/rebalancer-sim/src/MessageTracker.ts +++ /dev/null @@ -1,312 +0,0 @@ -import { ethers } from 'ethers'; -import { EventEmitter } from 'events'; - -import { MockMailbox__factory } from '@hyperlane-xyz/core'; -import { rootLogger } from '@hyperlane-xyz/utils'; - -import type { DeployedDomain } from './types.js'; - -const logger = rootLogger.child({ module: 'MessageTracker' }); - -/** - * Tracked message for off-chain processing control - */ -export interface TrackedMessage { - id: string; - transferId: string; - origin: string; - destination: string; - /** Nonce on the destination mailbox */ - destinationNonce: number; - /** When the message was dispatched */ - dispatchedAt: number; - /** When we should attempt delivery */ - deliveryTime: number; - /** Processing status */ - status: 'pending' | 'inflight' | 'delivered' | 'failed'; - /** Number of delivery attempts */ - attempts: number; - /** Last error if failed */ - lastError?: string; -} - -/** - * MessageTracker provides off-chain tracking and selective processing - * of Hyperlane messages. Fires transactions in parallel without blocking - * on receipts, similar to how the Hyperlane relayer batches messages. - */ -export class MessageTracker extends EventEmitter { - private messages: Map = new Map(); - private messageCounter = 0; - private destinationNonces: Map = new Map(); - private signer: ethers.Wallet; - private currentNonce: number = 0; - private nonceInitialized = false; - - constructor( - private readonly provider: ethers.providers.JsonRpcProvider, - private readonly domains: Record, - signerKey: string, - ) { - super(); - this.signer = new ethers.Wallet(signerKey, provider); - } - - /** - * Initialize by fetching current nonces from all destination mailboxes - */ - async initialize(): Promise { - for (const [chainName, domain] of Object.entries(this.domains)) { - const mailbox = MockMailbox__factory.connect( - domain.mailbox, - this.provider, - ); - const nonce = await mailbox.inboundUnprocessedNonce(); - this.destinationNonces.set(chainName, Number(nonce)); - } - // Initialize signer nonce for parallel tx submission - this.currentNonce = await this.signer.getTransactionCount(); - this.nonceInitialized = true; - } - - /** - * Track a new message after a transfer is initiated. - * Call this after transferRemote() succeeds. - */ - async trackMessage( - transferId: string, - origin: string, - destination: string, - deliveryDelay: number, - ): Promise { - const destDomain = this.domains[destination]; - const mailbox = MockMailbox__factory.connect( - destDomain.mailbox, - this.provider, - ); - await mailbox.inboundUnprocessedNonce(); // Verify mailbox is accessible - - const expectedNonce = this.destinationNonces.get(destination) || 0; - this.destinationNonces.set(destination, expectedNonce + 1); - - const message: TrackedMessage = { - id: `msg-${this.messageCounter++}`, - transferId, - origin, - destination, - destinationNonce: expectedNonce, - dispatchedAt: Date.now(), - deliveryTime: Date.now() + deliveryDelay, - status: 'pending', - attempts: 0, - }; - - this.messages.set(message.id, message); - this.emit('message_tracked', message); - - return message; - } - - /** - * Get all messages ready for delivery (past their delivery time, not inflight) - */ - getReadyMessages(): TrackedMessage[] { - const now = Date.now(); - return Array.from(this.messages.values()).filter( - (m) => m.status === 'pending' && m.deliveryTime <= now, - ); - } - - /** - * Get all pending messages (including not yet ready and inflight) - */ - getPendingMessages(): TrackedMessage[] { - return Array.from(this.messages.values()).filter( - (m) => m.status === 'pending' || m.status === 'inflight', - ); - } - - /** - * Process all ready messages in parallel without blocking on receipts. - * Fires transactions and subscribes to completion asynchronously. - */ - async processReadyMessages(): Promise<{ delivered: number; failed: number }> { - const ready = this.getReadyMessages(); - if (ready.length === 0) { - return { delivered: 0, failed: 0 }; - } - - // Ensure nonce is initialized - if (!this.nonceInitialized) { - this.currentNonce = await this.signer.getTransactionCount(); - this.nonceInitialized = true; - } - - // Check which messages can actually be processed (have sufficient liquidity) - // by doing a static call first - const processable: TrackedMessage[] = []; - - const checkStartTime = Date.now(); - for (const message of ready) { - const destDomain = this.domains[message.destination]; - const mailbox = MockMailbox__factory.connect( - destDomain.mailbox, - this.signer, - ); - - const staticCallStart = Date.now(); - try { - // Static call to check if it would succeed - await mailbox.callStatic.processInboundMessage( - message.destinationNonce, - ); - const staticCallDuration = Date.now() - staticCallStart; - if (staticCallDuration > 100) { - logger.warn( - { transferId: message.transferId, staticCallDuration }, - 'Slow static call', - ); - } - processable.push(message); - // Log successful processing after retries - if (message.attempts > 0) { - const waitTime = Date.now() - message.dispatchedAt; - logger.debug( - { - transferId: message.transferId, - origin: message.origin, - destination: message.destination, - attempts: message.attempts, - waitTime, - }, - 'Message ready after retries', - ); - } - } catch (error: any) { - const staticCallDuration = Date.now() - staticCallStart; - const errorMsg = error.reason || error.message || ''; - // Check if message was already delivered (e.g., by bridge controller) - // This is a permanent state, not a temporary error - if (errorMsg.includes('already delivered')) { - message.status = 'delivered'; - this.emit('message_delivered', message); - continue; - } - // Other errors - mark attempt but keep pending for retry - message.attempts++; - message.lastError = errorMsg; - - // Log failures - every 5 attempts or on slow static calls - if (message.attempts % 5 === 0 || staticCallDuration > 100) { - const waitTime = Date.now() - message.dispatchedAt; - logger.debug( - { - transferId: message.transferId, - origin: message.origin, - destination: message.destination, - attempts: message.attempts, - waitTime, - error: errorMsg, - }, - 'Message delivery failed, will retry', - ); - } - } - } - - const totalCheckTime = Date.now() - checkStartTime; - if (totalCheckTime > 500) { - logger.warn( - { messageCount: ready.length, totalCheckTime }, - 'Slow static call checks', - ); - } - - if (processable.length === 0) { - // No messages processable yet - not a failure, they will retry - return { delivered: 0, failed: 0 }; - } - - // Fire all processable transactions in parallel - const txPromises: Array<{ - message: TrackedMessage; - txPromise: Promise; - }> = []; - - for (const message of processable) { - message.status = 'inflight'; - message.attempts++; - - const destDomain = this.domains[message.destination]; - const mailbox = MockMailbox__factory.connect( - destDomain.mailbox, - this.signer, - ); - - // Fire transaction with explicit nonce (don't wait) - const txPromise = mailbox.processInboundMessage( - message.destinationNonce, - { nonce: this.currentNonce++ }, - ); - - txPromises.push({ message, txPromise }); - } - - // Subscribe to all tx completions asynchronously - let delivered = 0; - let failed = 0; - - await Promise.all( - txPromises.map(async ({ message, txPromise }) => { - try { - const tx = await txPromise; - await tx.wait(); - - message.status = 'delivered'; - this.emit('message_delivered', message); - delivered++; - } catch (error: any) { - // Transaction failed - back to pending for retry - message.status = 'pending'; - message.lastError = error.reason || error.message; - failed++; - } - }), - ); - - return { delivered, failed }; - } - - /** - * Check if there are any pending or inflight messages - */ - hasPendingMessages(): boolean { - return this.getPendingMessages().length > 0; - } - - /** - * Get message by transfer ID - */ - getMessageByTransferId(transferId: string): TrackedMessage | undefined { - return Array.from(this.messages.values()).find( - (m) => m.transferId === transferId, - ); - } - - /** - * Get all messages - */ - getAllMessages(): TrackedMessage[] { - return Array.from(this.messages.values()); - } - - /** - * Clear all tracked messages (for reset) - */ - clear(): void { - this.messages.clear(); - this.messageCounter = 0; - this.destinationNonces.clear(); - this.nonceInitialized = false; - } -} diff --git a/typescript/rebalancer-sim/src/MockInfrastructureController.ts b/typescript/rebalancer-sim/src/MockInfrastructureController.ts new file mode 100644 index 00000000000..357f408a2d9 --- /dev/null +++ b/typescript/rebalancer-sim/src/MockInfrastructureController.ts @@ -0,0 +1,346 @@ +import { ethers } from 'ethers'; + +import type { HyperlaneCore } from '@hyperlane-xyz/sdk'; +import { rootLogger } from '@hyperlane-xyz/utils'; + +import { KPICollector } from './KPICollector.js'; +import { MockActionTracker } from './runners/MockActionTracker.js'; +import type { + BridgeMockConfig, + BridgeRouteConfig, + DeployedDomain, +} from './types.js'; +import { DEFAULT_BRIDGE_ROUTE_CONFIG } from './types.js'; + +const logger = rootLogger.child({ module: 'MockInfrastructureController' }); + +/** Pending message awaiting delayed delivery */ +interface PendingMessage { + /** keccak256(message) — real Hyperlane messageId */ + messageId: string; + /** Full message bytes hex */ + message: string; + destination: string; + deliveryTime: number; + type: 'user-transfer' | 'bridge-transfer'; + /** Origin chain name */ + origin: string; + /** Decoded amount from body */ + amount: bigint; + /** Number of delivery attempts */ + attempts: number; +} + +/** + * MockInfrastructureController listens for Dispatch events on all Mailboxes, + * classifies messages by sender (warp vs bridge), and delivers them with + * configurable delays by calling process('0x', message) on the destination mailbox. + * + * Auto-tracks both user transfers and bridge transfers from Dispatch events — + * no external registration needed. + */ +export class MockInfrastructureController { + private pendingMessages: PendingMessage[] = []; + private signer!: ethers.Signer; + private isRunning = false; + private processing = false; + private processLoopTimer?: NodeJS.Timeout; + private currentNonce: number = 0; + private nonceInitialized = false; + + constructor( + private readonly core: HyperlaneCore, + private readonly domains: Record, + private readonly bridgeDelayConfig: BridgeMockConfig, + private readonly userTransferDelay: number, + private readonly kpiCollector: KPICollector, + private readonly actionTracker?: MockActionTracker, + ) {} + + private getRouteConfig( + origin: string, + destination: string, + ): BridgeRouteConfig { + return ( + this.bridgeDelayConfig[origin]?.[destination] ?? + DEFAULT_BRIDGE_ROUTE_CONFIG + ); + } + + private calculateBridgeDelay(config: BridgeRouteConfig): number { + const jitter = (Math.random() - 0.5) * 2 * config.deliveryJitter; + return Math.max(0, config.deliveryDelay + jitter); + } + + /** + * Start listening for Dispatch events and processing messages + */ + async start(): Promise { + if (this.isRunning) return; + this.isRunning = true; + + // Get signer from MultiProvider for nonce management + const firstChain = this.core.multiProvider.getKnownChainNames()[0]; + this.signer = this.core.multiProvider.getSigner(firstChain); + this.currentNonce = await this.signer.getTransactionCount(); + this.nonceInitialized = true; + + // Listen for Dispatch events on all mailboxes + for (const chainName of this.core.multiProvider.getKnownChainNames()) { + const mailbox = this.core.getContracts(chainName).mailbox; + mailbox.on( + mailbox.filters.Dispatch(), + ( + sender: string, + destination: number, + _recipient: string, + message: string, + ) => { + void this.onDispatch(chainName, sender, destination, message); + }, + ); + } + + // Start processing loop + this.processLoopTimer = setInterval(() => { + void this.processReadyMessages(); + }, 50); + } + + /** + * Handle a Dispatch event + */ + private async onDispatch( + originChain: string, + sender: string, + destinationDomainId: number, + message: string, + ): Promise { + const destChain = + this.core.multiProvider.tryGetChainName(destinationDomainId); + if (!destChain) { + logger.error({ destinationDomainId }, 'Unknown destination domain'); + return; + } + + const originDomain = this.domains[originChain]; + const senderLower = sender.toLowerCase(); + + // Classify by sender + const isWarp = senderLower === originDomain.warpToken.toLowerCase(); + const isBridge = senderLower === originDomain.bridge.toLowerCase(); + + if (!isWarp && !isBridge) { + logger.warn( + { sender, warp: originDomain.warpToken, bridge: originDomain.bridge }, + 'Unknown sender in Dispatch event', + ); + return; + } + + const type = isWarp ? 'user-transfer' : 'bridge-transfer'; + + // Compute real messageId + const messageId = ethers.utils.keccak256(message); + + // Decode amount from body (offset 77 bytes into message) + // TokenRouter encodes: TokenMessage.format(recipient, _outboundAmount(amount)) + // where _outboundAmount = amount * scale. Scale = 10^decimals for warp tokens. + // We divide by scale to recover the original wei amount. + const bodyOffset = 77; + const body = '0x' + message.slice(2 + bodyOffset * 2); + let amount = 0n; + try { + const decoded = ethers.utils.defaultAbiCoder.decode( + ['bytes32', 'uint256'], + body, + ); + const scaledAmount = decoded[1].toBigInt(); + // Warp tokens use scale = 10^18, bridge Router uses scale = 1 (no scaling) + amount = + type === 'user-transfer' ? scaledAmount / BigInt(1e18) : scaledAmount; + } catch (error) { + logger.warn( + { messageId, origin: originChain, dest: destChain, error }, + 'Failed to decode message amount', + ); + } + + // Calculate delivery time + let delay: number; + if (type === 'user-transfer') { + delay = this.userTransferDelay; + } else { + const routeConfig = this.getRouteConfig(originChain, destChain); + delay = this.calculateBridgeDelay(routeConfig); + } + + const pending: PendingMessage = { + messageId, + message, + destination: destChain, + deliveryTime: Date.now() + delay, + type, + origin: originChain, + amount, + attempts: 0, + }; + + if (type === 'bridge-transfer') { + // Record rebalance start in KPI + const rebalanceId = this.kpiCollector.recordRebalanceStart( + originChain, + destChain, + amount, + 0n, + ); + this.kpiCollector.linkBridgeTransfer(messageId, rebalanceId); + } else { + // User transfer: auto-track from event + this.kpiCollector.recordTransferStart( + messageId, + originChain, + destChain, + amount, + ); + + this.actionTracker?.addTransfer( + messageId, + this.core.multiProvider.getDomainId(originChain), + this.core.multiProvider.getDomainId(destChain), + amount, + ); + } + + this.pendingMessages.push(pending); + } + + /** + * Process messages that are past their delivery time. + * Retries indefinitely — waitForAllDeliveries handles the timeout. + */ + private async processReadyMessages(): Promise { + if (this.processing) return; + this.processing = true; + try { + await this.doProcessReadyMessages(); + } finally { + this.processing = false; + } + } + + private async doProcessReadyMessages(): Promise { + const now = Date.now(); + const ready = this.pendingMessages.filter((m) => m.deliveryTime <= now); + if (ready.length === 0) return; + + if (!this.nonceInitialized) { + this.currentNonce = await this.signer.getTransactionCount(); + this.nonceInitialized = true; + } + + for (const msg of ready) { + const mailbox = this.core.getContracts(msg.destination).mailbox; + + // Static call pre-check + try { + await mailbox.callStatic.process('0x', msg.message); + } catch { + msg.attempts++; + msg.deliveryTime = now + 200; + continue; + } + + try { + const tx = await mailbox.process('0x', msg.message, { + nonce: this.currentNonce++, + }); + await tx.wait(); + + // Remove from pending + const idx = this.pendingMessages.indexOf(msg); + if (idx >= 0) this.pendingMessages.splice(idx, 1); + + // Record completion + if (msg.type === 'user-transfer') { + this.kpiCollector.recordTransferComplete(msg.messageId); + this.actionTracker?.removeTransfer(msg.messageId); + } else if (msg.type === 'bridge-transfer') { + this.kpiCollector.recordRebalanceComplete(msg.messageId); + if (this.actionTracker && msg.amount) { + this.actionTracker.completeRebalanceByRoute( + this.core.multiProvider.getDomainId(msg.origin), + this.core.multiProvider.getDomainId(msg.destination), + msg.amount, + ); + } + } + } catch (error) { + msg.attempts++; + msg.deliveryTime = now + 200; + // Re-sync nonce on tx failure + this.currentNonce = await this.signer.getTransactionCount(); + logger.debug( + { messageId: msg.messageId, dest: msg.destination, error }, + 'Delivery tx failed, will retry', + ); + } + } + } + + /** + * Stop listening and processing + */ + async stop(): Promise { + this.isRunning = false; + + if (this.processLoopTimer) { + clearInterval(this.processLoopTimer); + this.processLoopTimer = undefined; + } + + for (const chainName of this.core.multiProvider.getKnownChainNames()) { + this.core.getContracts(chainName).mailbox.removeAllListeners(); + } + } + + hasPendingMessages(): boolean { + return this.pendingMessages.length > 0; + } + + /** + * Wait for all pending messages to be delivered + */ + async waitForAllDeliveries(timeoutMs: number = 30000): Promise { + const startTime = Date.now(); + + while (this.hasPendingMessages()) { + if (Date.now() - startTime > timeoutMs) { + const remaining = this.pendingMessages.length; + logger.warn( + { remaining }, + 'Timeout waiting for deliveries - marking failures', + ); + + for (const msg of this.pendingMessages) { + if (msg.type === 'user-transfer') { + this.kpiCollector.recordTransferFailed(msg.messageId); + this.actionTracker?.removeTransfer(msg.messageId); + } else if (msg.type === 'bridge-transfer') { + this.kpiCollector.recordRebalanceFailed(msg.messageId); + if (this.actionTracker && msg.amount) { + this.actionTracker.completeRebalanceByRoute( + this.core.multiProvider.getDomainId(msg.origin), + this.core.multiProvider.getDomainId(msg.destination), + msg.amount, + ); + } + } + } + this.pendingMessages = []; + break; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + } +} diff --git a/typescript/rebalancer-sim/src/RebalancerSimulationHarness.ts b/typescript/rebalancer-sim/src/RebalancerSimulationHarness.ts index 871e54cbd97..9b1f451b63e 100644 --- a/typescript/rebalancer-sim/src/RebalancerSimulationHarness.ts +++ b/typescript/rebalancer-sim/src/RebalancerSimulationHarness.ts @@ -239,15 +239,6 @@ export class RebalancerSimulationHarness { return this.deployment; } - /** - * Reset the simulation engine state (does not reset blockchain state) - */ - reset(): void { - if (this.engine) { - this.engine.reset(); - } - } - /** * Generate a markdown report from simulation results */ diff --git a/typescript/rebalancer-sim/src/SimulationDeployment.ts b/typescript/rebalancer-sim/src/SimulationDeployment.ts index 6642e423ed4..091cba633ee 100644 --- a/typescript/rebalancer-sim/src/SimulationDeployment.ts +++ b/typescript/rebalancer-sim/src/SimulationDeployment.ts @@ -150,16 +150,43 @@ export async function deployMultiDomainSimulation( await warpToken.enrollRemoteRouters(remoteDomains, remoteRouters); } - // Step 6: Deploy MockValueTransferBridge for each domain and add to allowed bridges + // Step 6: Deploy MockValueTransferBridge for each domain (now extends Router) const bridges: Record = {}; for (const chain of chains) { const bridge = await new MockValueTransferBridge__factory(deployer).deploy( collateralTokens[chain.domainId].address, + mailboxes[chain.domainId].address, ); await bridge.deployed(); + + // Initialize the bridge (Router requires initialization) + await bridge.initialize( + ethers.constants.AddressZero, // hook + ethers.constants.AddressZero, // ISM + deployerAddress, // owner + ); + bridges[chain.domainId] = bridge; } + // Step 6b: Enroll remote routers on bridges (so _Router_dispatch works) + for (const chain of chains) { + const bridge = bridges[chain.domainId]; + const remoteDomains: number[] = []; + const remoteRouters: string[] = []; + + for (const otherChain of chains) { + if (chain.domainId !== otherChain.domainId) { + remoteDomains.push(otherChain.domainId); + remoteRouters.push( + ethers.utils.hexZeroPad(bridges[otherChain.domainId].address, 32), + ); + } + } + + await bridge.enrollRemoteRouters(remoteDomains, remoteRouters); + } + // Step 7: Add bridges to warp tokens for all destination domains for (const chain of chains) { const warpToken = warpTokens[chain.domainId]; diff --git a/typescript/rebalancer-sim/src/SimulationEngine.ts b/typescript/rebalancer-sim/src/SimulationEngine.ts index 231858030f3..24ce151e7a6 100644 --- a/typescript/rebalancer-sim/src/SimulationEngine.ts +++ b/typescript/rebalancer-sim/src/SimulationEngine.ts @@ -4,16 +4,19 @@ import { ERC20__factory, HypERC20Collateral__factory, } from '@hyperlane-xyz/core'; -import { TokenStandard, type WarpCoreConfig } from '@hyperlane-xyz/sdk'; -import { rootLogger } from '@hyperlane-xyz/utils'; +import { + HyperlaneCore, + MultiProvider, + TokenStandard, + type WarpCoreConfig, +} from '@hyperlane-xyz/sdk'; +import { ProtocolType, rootLogger } from '@hyperlane-xyz/utils'; -import { BridgeMockController } from './BridgeMockController.js'; import { KPICollector } from './KPICollector.js'; -import { MessageTracker } from './MessageTracker.js'; +import { MockInfrastructureController } from './MockInfrastructureController.js'; import type { BridgeMockConfig, IRebalancerRunner, - InflightContextCallbacks, MultiDomainDeploymentResult, RebalancerSimConfig, SimulationResult, @@ -38,13 +41,7 @@ export const DEFAULT_TIMING: SimulationTiming = { */ export class SimulationEngine { private provider: ethers.providers.JsonRpcProvider; - private bridgeController?: BridgeMockController; - private kpiCollector?: KPICollector; - private messageTracker?: MessageTracker; private isRunning = false; - private mailboxProcessingInterval?: NodeJS.Timeout; - private mailboxProcessingInFlight = false; - private inflightCallbacks?: InflightContextCallbacks; constructor(private readonly deployment: MultiDomainDeploymentResult) { this.provider = new ethers.providers.JsonRpcProvider(deployment.anvilRpc); @@ -67,86 +64,32 @@ export class SimulationEngine { const startTime = Date.now(); this.isRunning = true; + let controller: MockInfrastructureController | undefined; + try { - // Initialize components - // Use bridgeControllerKey for bridge operations to avoid nonce conflicts - this.bridgeController = new BridgeMockController( + // Initialize KPI collector + const kpiCollector = new KPICollector( this.provider, this.deployment.domains, - this.deployment.bridgeControllerKey, - bridgeConfig, ); + await kpiCollector.initialize(); - this.kpiCollector = new KPICollector( - this.provider, - this.deployment.domains, - ); + // Get action tracker from rebalancer if supported + const actionTracker = rebalancer.getActionTracker?.(); - // Initialize MessageTracker for off-chain message tracking - this.messageTracker = new MessageTracker( - this.provider, + // Build HyperlaneCore for the controller (manages MultiProvider + Mailboxes) + const core = this.buildHyperlaneCore(); + + // Create unified controller + controller = new MockInfrastructureController( + core, this.deployment.domains, - this.deployment.mailboxProcessorKey, + bridgeConfig, + timing.userTransferDeliveryDelay, + kpiCollector, + actionTracker, ); - - await this.kpiCollector.initialize(); - await this.messageTracker.initialize(); - await this.bridgeController.start(); - - // Wire up MessageTracker events for KPI tracking - this.messageTracker.on('message_delivered', (message) => { - this.kpiCollector!.recordTransferComplete(message.transferId); - - // Notify inflight context adapter - this.inflightCallbacks?.onTransferDelivered(message.transferId); - }); - - this.messageTracker.on('message_failed', ({ message }) => { - // Don't record as failed yet - it will retry - logger.debug( - { - messageId: message.id, - attempts: message.attempts, - error: message.lastError, - }, - 'Message failed, will retry', - ); - }); - - // Set up bridge event handlers for rebalance KPI tracking - // Bridge transfers are rebalancer operations (user transfers go through warp token) - this.bridgeController.on('transfer_initiated', (event) => { - const rebalanceId = this.kpiCollector!.recordRebalanceStart( - event.transfer.origin, - event.transfer.destination, - event.transfer.amount, - BigInt(0), // Gas cost not tracked yet - ); - // Link bridge transfer ID to rebalance ID for completion tracking - this.kpiCollector!.linkBridgeTransfer(event.transfer.id, rebalanceId); - - // Notify inflight context adapter - this.inflightCallbacks?.onRebalanceInitiated( - event.transfer.id, - event.transfer.origin, - event.transfer.destination, - event.transfer.amount, - ); - }); - - this.bridgeController.on('transfer_delivered', (event) => { - this.kpiCollector!.recordRebalanceComplete(event.transfer.id); - - // Notify inflight context adapter - this.inflightCallbacks?.onRebalanceDelivered(event.transfer.id); - }); - - this.bridgeController.on('transfer_failed', (event) => { - this.kpiCollector!.recordRebalanceFailed(event.transfer.id); - - // Notify inflight context adapter (treat failed as delivered for inflight tracking) - this.inflightCallbacks?.onRebalanceDelivered(event.transfer.id); - }); + await controller.start(); // Build warp config for rebalancer const warpConfig = this.buildWarpConfig(); @@ -160,34 +103,22 @@ export class SimulationEngine { }; await rebalancer.initialize(rebalancerConfig); - - // Get inflight callbacks if the rebalancer supports them - this.inflightCallbacks = rebalancer.getInflightCallbacks?.(); - - // Start rebalancer daemon await rebalancer.start(); - // Start periodic mailbox processing for delayed user transfer delivery - this.startMailboxProcessing(); - // Execute transfers according to scenario - await this.executeTransfers(scenario, timing); + await this.executeTransfers(scenario, timing, kpiCollector); - // Wait for all user transfer deliveries (respecting delay) - // Use a timeout to prevent indefinite hanging - await Promise.race([ - this.waitForUserTransferDeliveries(), - new Promise((resolve) => setTimeout(resolve, 60000)), // 60s max - ]); + // Wait for ethers event polling to catch up + await new Promise((r) => setTimeout(r, 200)); - // Wait for bridge deliveries to complete (rebalancer transfers) - await this.bridgeController.waitForAllDeliveries(30000); + // Wait for all deliveries (user transfers + bridge transfers) + await controller.waitForAllDeliveries(60000); // Wait for rebalancer to become idle await rebalancer.waitForIdle(5000); // Generate final KPIs - const kpis = await this.kpiCollector.generateKPIs(); + const kpis = await kpiCollector.generateKPIs(); const endTime = Date.now(); return { @@ -197,13 +128,11 @@ export class SimulationEngine { endTime, duration: endTime - startTime, kpis, - transferRecords: this.kpiCollector.getTransferRecords(), - rebalanceRecords: this.kpiCollector.getRebalanceRecords(), + transferRecords: kpiCollector.getTransferRecords(), + rebalanceRecords: kpiCollector.getRebalanceRecords(), }; } finally { - // Always cleanup, even if we timeout or error this.isRunning = false; - this.stopMailboxProcessing(); try { await rebalancer.stop(); @@ -211,24 +140,17 @@ export class SimulationEngine { // Ignore stop errors } - if (this.bridgeController) { + if (controller) { try { - await this.bridgeController.stop(); + await controller.stop(); } catch { // Ignore stop errors } } - if (this.messageTracker) { - this.messageTracker.removeAllListeners(); - } - // Clean up provider to release connections this.provider.removeAllListeners(); - // Force polling to stop this.provider.polling = false; - // Clear inflight callbacks - this.inflightCallbacks = undefined; } } @@ -238,6 +160,7 @@ export class SimulationEngine { private async executeTransfers( scenario: TransferScenario, timing: SimulationTiming, + kpiCollector: KPICollector, ): Promise { const deployer = new ethers.Wallet( this.deployment.deployerKey, @@ -255,14 +178,6 @@ export class SimulationEngine { await new Promise((resolve) => setTimeout(resolve, waitTime)); } - // Record transfer start - this.kpiCollector!.recordTransferStart( - transfer.id, - transfer.origin, - transfer.destination, - transfer.amount, - ); - // Execute the transfer via warp token const originDomain = this.deployment.domains[transfer.origin]; const destDomain = this.deployment.domains[transfer.destination]; @@ -311,21 +226,7 @@ export class SimulationEngine { ); } - // Track message for delayed delivery via MessageTracker - await this.messageTracker!.trackMessage( - transfer.id, - transfer.origin, - transfer.destination, - timing.userTransferDeliveryDelay, - ); - - // Notify inflight context adapter that a user transfer is pending - this.inflightCallbacks?.onTransferInitiated( - transfer.id, - transfer.origin, - transfer.destination, - transfer.amount, - ); + // Controller auto-tracks from Dispatch events — no registration needed } catch (error) { logger.error( { @@ -334,91 +235,16 @@ export class SimulationEngine { }, 'Transfer failed', ); - this.kpiCollector!.recordTransferFailed(transfer.id); - } - } - logger.info('All transfers executed'); - } - - /** - * Start periodic processing of mailbox messages (simulates relayer with delay) - */ - private startMailboxProcessing(): void { - // Process mailbox every 100ms to check for deliveries due - const PROCESS_INTERVAL = 100; - - this.mailboxProcessingInterval = setInterval(async () => { - // Guard against overlapping ticks to prevent nonce collisions - if (this.mailboxProcessingInFlight) return; - this.mailboxProcessingInFlight = true; - try { - await this.processReadyMailboxDeliveries(); - } finally { - this.mailboxProcessingInFlight = false; - } - }, PROCESS_INTERVAL); - } - - /** - * Stop mailbox processing - */ - private stopMailboxProcessing(): void { - if (this.mailboxProcessingInterval) { - clearInterval(this.mailboxProcessingInterval); - this.mailboxProcessingInterval = undefined; - } - } - - /** - * Process mailbox deliveries that are ready (past their delivery time) - * Uses MessageTracker for off-chain tracking with per-message control - */ - private async processReadyMailboxDeliveries(): Promise { - if (!this.messageTracker) return; - await this.messageTracker.processReadyMessages(); - } - - /** - * Wait for all pending user transfer deliveries to complete - */ - private async waitForUserTransferDeliveries( - timeout: number = 30000, - ): Promise { - if (!this.messageTracker) return; - - const startTime = Date.now(); - - while (this.messageTracker.hasPendingMessages()) { - if (Date.now() - startTime > timeout) { - const pending = this.messageTracker.getPendingMessages(); - logger.warn( - { pendingCount: pending.length }, - 'Timeout waiting for user transfer deliveries - marking as failed', + kpiCollector.recordTransferStart( + transfer.id, + transfer.origin, + transfer.destination, + transfer.amount, ); - // Mark pending messages as failed so KPIs reflect reality - for (const msg of pending) { - logger.warn( - { - messageId: msg.id, - origin: msg.origin, - destination: msg.destination, - status: msg.status, - attempts: msg.attempts, - error: msg.lastError || 'timeout', - }, - 'Marking pending message as failed', - ); - // Record as failed in KPI collector - this.kpiCollector?.recordTransferFailed(msg.transferId); - } - // Clear pending messages so they don't block - this.messageTracker.clear(); - break; + kpiCollector.recordTransferFailed(transfer.id); } - - // Interval handles processing; just wait - await new Promise((resolve) => setTimeout(resolve, 100)); } + logger.info('All transfers executed'); } /** @@ -446,13 +272,40 @@ export class SimulationEngine { } /** - * Reset internal tracking state (does not reset blockchain state) + * Build a HyperlaneCore for the infrastructure controller with the + * mailbox processor signer set on all chains. */ - reset(): void { - // Clear message tracker state - if (this.messageTracker) { - this.messageTracker.clear(); + private buildHyperlaneCore(): HyperlaneCore { + const chainMetadata: Record = {}; + const addressesMap: Record = {}; + for (const [chainName, domain] of Object.entries(this.deployment.domains)) { + chainMetadata[chainName] = { + name: chainName, + chainId: 31337, + domainId: domain.domainId, + protocol: ProtocolType.Ethereum, + rpcUrls: [{ http: this.deployment.anvilRpc }], + nativeToken: { name: 'Ether', symbol: 'ETH', decimals: 18 }, + }; + addressesMap[chainName] = { mailbox: domain.mailbox }; + } + + const multiProvider = new MultiProvider(chainMetadata); + const processorWallet = new ethers.Wallet( + this.deployment.mailboxProcessorKey, + this.provider, + ); + multiProvider.setSharedSigner(processorWallet); + + // Set fast polling on internal providers + for (const chainName of multiProvider.getKnownChainNames()) { + const p = multiProvider.tryGetProvider(chainName); + if (p && 'pollingInterval' in p) { + (p as ethers.providers.JsonRpcProvider).pollingInterval = 100; + } } + + return HyperlaneCore.fromAddressesMap(addressesMap, multiProvider); } /** diff --git a/typescript/rebalancer-sim/src/index.ts b/typescript/rebalancer-sim/src/index.ts index 9daa8d598e3..ada8749a5c1 100644 --- a/typescript/rebalancer-sim/src/index.ts +++ b/typescript/rebalancer-sim/src/index.ts @@ -6,9 +6,8 @@ */ // Core simulation classes -export { BridgeMockController } from './BridgeMockController.js'; export { KPICollector } from './KPICollector.js'; -export { MessageTracker } from './MessageTracker.js'; +export { MockInfrastructureController } from './MockInfrastructureController.js'; export { RebalancerSimulationHarness } from './RebalancerSimulationHarness.js'; export { deployMultiDomainSimulation, diff --git a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts index 06d8fa24262..0cc39b87bc6 100644 --- a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts +++ b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts @@ -209,19 +209,9 @@ export class MockActionTracker implements IActionTracker { async failRebalanceIntent(id: string): Promise { const intent = this.intents.get(id); if (intent) { - // In simulation, RebalancerService calls failRebalanceIntent when no Dispatch event - // is found (MockValueTransferBridge doesn't emit Dispatch). But the bridge transfer - // DID succeed - we'll see the SentTransferRemote event. So we keep the intent as - // in_progress to allow createActionForPendingIntent to work when the event fires. - // - // Note: This is a simulation-specific behavior. In production, no Dispatch event - // means the transfer actually failed. - intent.status = 'in_progress'; + intent.status = 'failed'; intent.updatedAt = Date.now(); - logger.debug( - { id, originalIntentAmount: intent.amount.toString() }, - 'Rebalance intent "failed" - keeping as in_progress for simulation (bridge transfer succeeded)', - ); + logger.debug({ id }, 'Rebalance intent failed'); } } diff --git a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts index b33c73e3387..8b6f2c55070 100644 --- a/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts +++ b/typescript/rebalancer-sim/src/runners/ProductionRebalancerRunner.ts @@ -11,11 +11,7 @@ import type { StrategyConfig } from '@hyperlane-xyz/rebalancer'; import { MultiProtocolProvider, MultiProvider } from '@hyperlane-xyz/sdk'; import { ProtocolType, rootLogger } from '@hyperlane-xyz/utils'; -import type { - IRebalancerRunner, - InflightContextCallbacks, - RebalancerSimConfig, -} from '../types.js'; +import type { IRebalancerRunner, RebalancerSimConfig } from '../types.js'; import { MockActionTracker } from './MockActionTracker.js'; import { SimulationRegistry } from './SimulationRegistry.js'; @@ -108,15 +104,6 @@ function buildStrategyConfig(config: RebalancerSimConfig): StrategyConfig { * ProductionRebalancerRunner runs the actual RebalancerService in-process. * This wraps the real CLI rebalancer for simulation testing. */ -/** - * Tracks pending rebalance info for correlating bridge callbacks with tracker actions. - */ -interface PendingRebalanceInfo { - origin: number; // domain ID - destination: number; // domain ID - amount: bigint; -} - export class ProductionRebalancerRunner extends EventEmitter implements IRebalancerRunner @@ -126,9 +113,7 @@ export class ProductionRebalancerRunner private config?: RebalancerSimConfig; private service?: RebalancerService; private running = false; - private mockTracker?: MockActionTracker; - // Maps bridge transfer ID to rebalance info for delivery correlation - private pendingRebalances = new Map(); + private mockTracker = new MockActionTracker(); async initialize(config: RebalancerSimConfig): Promise { // Cleanup any previously running instance @@ -136,8 +121,8 @@ export class ProductionRebalancerRunner this.config = config; - // Create mock action tracker for simulation - this.mockTracker = new MockActionTracker(); + // Reset tracker state for fresh simulation + this.mockTracker.clear(); } async start(): Promise { @@ -295,11 +280,7 @@ export class ProductionRebalancerRunner } this.config = undefined; - if (this.mockTracker) { - this.mockTracker.clear(); - this.mockTracker = undefined; - } - this.pendingRebalances.clear(); + this.mockTracker.clear(); this.removeAllListeners(); } @@ -314,78 +295,9 @@ export class ProductionRebalancerRunner } /** - * Get callbacks for wiring inflight context updates. - * SimulationEngine uses these to notify about pending transfers and rebalances. + * Get the mock action tracker for direct inflight tracking updates. */ - getInflightCallbacks(): InflightContextCallbacks | undefined { - if (!this.mockTracker || !this.config) { - return undefined; - } - - // Helper to convert chain name to domain ID - const getDomainId = (chainName: string): number => { - const domain = this.config!.deployment.domains[chainName]; - if (!domain) { - throw new Error(`Unknown chain: ${chainName}`); - } - return domain.domainId; - }; - - return { - onTransferInitiated: ( - id: string, - origin: string, - destination: string, - amount: bigint, - ) => { - this.mockTracker!.addTransfer( - id, - getDomainId(origin), - getDomainId(destination), - amount, - ); - }, - onTransferDelivered: (id: string) => { - this.mockTracker!.removeTransfer(id); - }, - onRebalanceInitiated: ( - id: string, - origin: string, - destination: string, - amount: bigint, - ) => { - const originDomain = getDomainId(origin); - const destDomain = getDomainId(destination); - - // Store mapping from bridge transfer ID to route info for delivery correlation - this.pendingRebalances.set(id, { - origin: originDomain, - destination: destDomain, - amount, - }); - - // Create action for the pending intent. - // RebalancerService can't extract messageId from MockValueTransferBridge - // (no Dispatch event), so we create the action ourselves. - this.mockTracker!.createActionForPendingIntent( - originDomain, - destDomain, - amount, - id, - ); - }, - onRebalanceDelivered: (id: string) => { - // Look up the route info and complete the matching action in tracker - const info = this.pendingRebalances.get(id); - if (info) { - this.pendingRebalances.delete(id); - this.mockTracker!.completeRebalanceByRoute( - info.origin, - info.destination, - info.amount, - ); - } - }, - }; + getActionTracker(): MockActionTracker { + return this.mockTracker; } } diff --git a/typescript/rebalancer-sim/src/types.ts b/typescript/rebalancer-sim/src/types.ts index 9150ed682f3..efd693540d0 100644 --- a/typescript/rebalancer-sim/src/types.ts +++ b/typescript/rebalancer-sim/src/types.ts @@ -7,6 +7,8 @@ import type { WarpCoreConfig } from '@hyperlane-xyz/sdk'; import type { Address } from '@hyperlane-xyz/utils'; +import type { MockActionTracker } from './runners/MockActionTracker.js'; + // ============================================================================= // BRIDGE TYPES // ============================================================================= @@ -377,43 +379,6 @@ export interface ChainStrategyConfig { bridgeLockTime: number; } -/** - * Callbacks for wiring inflight context updates in simulation. - * The simulation engine uses these to notify the rebalancer about - * pending transfers and rebalances. - */ -export interface InflightContextCallbacks { - /** - * Called when a user transfer is initiated (before delivery) - */ - onTransferInitiated: ( - id: string, - origin: string, - destination: string, - amount: bigint, - ) => void; - - /** - * Called when a user transfer is delivered - */ - onTransferDelivered: (id: string) => void; - - /** - * Called when a rebalance transfer is initiated via bridge - */ - onRebalanceInitiated: ( - id: string, - origin: string, - destination: string, - amount: bigint, - ) => void; - - /** - * Called when a rebalance transfer completes via bridge - */ - onRebalanceDelivered: (id: string) => void; -} - /** * Interface for rebalancer runners in simulation */ @@ -452,11 +417,11 @@ export interface IRebalancerRunner { on(event: 'rebalance', listener: (e: RebalancerEvent) => void): this; /** - * Get callbacks for inflight context updates (optional). - * If supported, SimulationEngine wires these to bridge/message events. - * Returns undefined if the runner doesn't support inflight tracking. + * Get the mock action tracker (optional). + * If supported, SimulationEngine passes it to the infrastructure controller + * for direct inflight tracking updates. */ - getInflightCallbacks?(): InflightContextCallbacks | undefined; + getActionTracker?(): MockActionTracker | undefined; } /** diff --git a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts index c8d5cebceb6..e6b0be26aba 100644 --- a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts +++ b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts @@ -227,7 +227,6 @@ describe('Rebalancer Simulation', function () { * significantly reduces redundant rebalances (typically 60-80% fewer) */ it('inflight-guard: ProductionRebalancer uses fewer rebalances with inflight tracking', async function () { - // This test takes longer due to 3s bridge delays and runs multiple rebalancers this.timeout(120000); const { results } = await runScenarioWithRebalancers('inflight-guard', { @@ -351,31 +350,4 @@ describe('Rebalancer Simulation', function () { ); } }); - - // ============================================================================ - // BASELINE (NO REBALANCER) - // ============================================================================ - - it('no-rebalancer baseline: shows transfer failures without rebalancing (report only)', async function () { - // Run with NoOpRebalancer to demonstrate what happens without active rebalancing - // This is report-only - no assertions, just generates visualization - const { results } = await runScenarioWithRebalancers( - 'extreme-drain-chain1', - { - anvilRpc: anvil.rpc, - rebalancerTypes: ['noop'], - }, - ); - - // Report only - log results but don't assert - const result = results[0]; - console.log(`\n NO-REBALANCER BASELINE (report only):`); - console.log( - ` Without rebalancing, completion rate: ${(result.kpis.completionRate * 100).toFixed(1)}%`, - ); - console.log( - ` Failed transfers: ${result.kpis.totalTransfers - result.kpis.completedTransfers}`, - ); - console.log(` Rebalances: ${result.kpis.totalRebalances} (expected: 0)`); - }); }); From bb4aedd845485ec3fd53f36e82a9115e81cfa929 Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 09:55:15 -0500 Subject: [PATCH 06/12] fix(rebalancer): Call initialize() on externally provided ActionTracker Co-Authored-By: Claude Opus 4.6 --- typescript/rebalancer/src/core/RebalancerService.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/typescript/rebalancer/src/core/RebalancerService.ts b/typescript/rebalancer/src/core/RebalancerService.ts index 05a73dd7951..8d8bfa8af84 100644 --- a/typescript/rebalancer/src/core/RebalancerService.ts +++ b/typescript/rebalancer/src/core/RebalancerService.ts @@ -194,6 +194,7 @@ export class RebalancerService { this.actionTracker, this.multiProvider, ); + await this.actionTracker.initialize(); this.logger.info('Using externally provided ActionTracker'); } else { const { tracker, adapter } = From 31e92f56510cdb28875b364f9a1475ad0d626dc0 Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 10:43:29 -0500 Subject: [PATCH 07/12] fix(rebalancer-sim): Address PR review feedback - Log static-call pre-check failures instead of silently swallowing - Use explicit `> 0n` check instead of falsy check for bigint amount - Add failRebalanceByRoute to MockActionTracker for timeout path (timed-out transfers should not be marked as complete) - Auto-complete parent intent in completeRebalanceAction - Add division-by-zero guard in inflight-guard test - Call initialize() on MockValueTransferBridge in e2e tests Co-Authored-By: Claude Opus 4.6 --- .../ethereum/warp/warp-rebalancer.e2e-test.ts | 20 ++++++++ .../src/MockInfrastructureController.ts | 17 +++++-- .../src/runners/MockActionTracker.ts | 47 +++++++++++++++++++ .../test/integration/full-simulation.test.ts | 4 ++ 4 files changed, 84 insertions(+), 4 deletions(-) diff --git a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts index 5b4dcf7c0a7..d8c771c7829 100644 --- a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts +++ b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts @@ -824,6 +824,11 @@ describe('hyperlane warp rebalancer e2e tests', async function () { const bridgeContract = await new MockValueTransferBridge__factory( chain3Signer, ).deploy(tokenChain3.address, chain3Addresses.mailbox); + await bridgeContract.initialize( + ethers.constants.AddressZero, + ethers.constants.AddressZero, + ANVIL_DEPLOYER_ADDRESS, + ); // Allow bridge await chain3CollateralContract.addBridge( @@ -878,6 +883,11 @@ describe('hyperlane warp rebalancer e2e tests', async function () { const bridgeContract = await new MockValueTransferBridge__factory( chain3Signer, ).deploy(tokenChain3.address, chain3Addresses.mailbox); + await bridgeContract.initialize( + ethers.constants.AddressZero, + ethers.constants.AddressZero, + ANVIL_DEPLOYER_ADDRESS, + ); // Allow bridge await chain3CollateralContract.addBridge( @@ -962,6 +972,11 @@ describe('hyperlane warp rebalancer e2e tests', async function () { const bridgeContract = await new MockValueTransferBridge__factory( originSigner, ).deploy(tokenChain3.address, chain3Addresses.mailbox); + await bridgeContract.initialize( + ethers.constants.AddressZero, + ethers.constants.AddressZero, + ANVIL_DEPLOYER_ADDRESS, + ); // Allow bridge // This allow the bridge to be used to send the rebalance transaction @@ -1209,6 +1224,11 @@ describe('hyperlane warp rebalancer e2e tests', async function () { const bridgeContract = await new MockValueTransferBridge__factory( originSigner, ).deploy(tokenChain3.address, chain3Addresses.mailbox); + await bridgeContract.initialize( + ethers.constants.AddressZero, + ethers.constants.AddressZero, + ANVIL_DEPLOYER_ADDRESS, + ); // Allow bridge // This allow the bridge to be used to send the rebalance transaction diff --git a/typescript/rebalancer-sim/src/MockInfrastructureController.ts b/typescript/rebalancer-sim/src/MockInfrastructureController.ts index 357f408a2d9..32da8c12e18 100644 --- a/typescript/rebalancer-sim/src/MockInfrastructureController.ts +++ b/typescript/rebalancer-sim/src/MockInfrastructureController.ts @@ -245,7 +245,16 @@ export class MockInfrastructureController { // Static call pre-check try { await mailbox.callStatic.process('0x', msg.message); - } catch { + } catch (error) { + logger.debug( + { + messageId: msg.messageId, + dest: msg.destination, + attempts: msg.attempts, + error, + }, + 'Static pre-check failed, will retry', + ); msg.attempts++; msg.deliveryTime = now + 200; continue; @@ -267,7 +276,7 @@ export class MockInfrastructureController { this.actionTracker?.removeTransfer(msg.messageId); } else if (msg.type === 'bridge-transfer') { this.kpiCollector.recordRebalanceComplete(msg.messageId); - if (this.actionTracker && msg.amount) { + if (this.actionTracker && msg.amount > 0n) { this.actionTracker.completeRebalanceByRoute( this.core.multiProvider.getDomainId(msg.origin), this.core.multiProvider.getDomainId(msg.destination), @@ -328,8 +337,8 @@ export class MockInfrastructureController { this.actionTracker?.removeTransfer(msg.messageId); } else if (msg.type === 'bridge-transfer') { this.kpiCollector.recordRebalanceFailed(msg.messageId); - if (this.actionTracker && msg.amount) { - this.actionTracker.completeRebalanceByRoute( + if (this.actionTracker && msg.amount > 0n) { + this.actionTracker.failRebalanceByRoute( this.core.multiProvider.getDomainId(msg.origin), this.core.multiProvider.getDomainId(msg.destination), msg.amount, diff --git a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts index 0cc39b87bc6..1859ddfc77a 100644 --- a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts +++ b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts @@ -268,6 +268,10 @@ export class MockActionTracker implements IActionTracker { if (intent) { intent.fulfilledAmount += action.amount; intent.updatedAt = Date.now(); + + if (intent.fulfilledAmount >= intent.amount) { + intent.status = 'complete'; + } } logger.debug({ id }, 'Rebalance action completed'); @@ -439,4 +443,47 @@ export class MockActionTracker implements IActionTracker { ); return true; } + + /** + * Fail a rebalance action by route match (origin, destination, amount). + * Marks the action as failed and the parent intent as failed. + * Does NOT increment fulfilledAmount. + */ + failRebalanceByRoute( + origin: Domain, + destination: Domain, + amount: bigint, + ): boolean { + const allActions = Array.from(this.actions.values()); + const matchingActions = allActions + .filter( + (a) => + a.origin === origin && + a.destination === destination && + a.amount === amount && + a.status === 'in_progress', + ) + .sort((a, b) => a.createdAt - b.createdAt); + + if (matchingActions.length === 0) { + return false; + } + + const action = matchingActions[0]; + action.status = 'failed'; + action.updatedAt = Date.now(); + + // Mark parent intent as failed + const intent = this.intents.get(action.intentId); + if (intent) { + intent.status = 'failed'; + intent.updatedAt = Date.now(); + } + + logger.debug( + { actionId: action.id, intentId: action.intentId, origin, destination }, + 'Failed rebalance action by route', + ); + return true; + } } diff --git a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts index e6b0be26aba..02d98761944 100644 --- a/typescript/rebalancer-sim/test/integration/full-simulation.test.ts +++ b/typescript/rebalancer-sim/test/integration/full-simulation.test.ts @@ -272,6 +272,10 @@ describe('Rebalancer Simulation', function () { // ProductionRebalancer should use at most 50% of SimpleRebalancer's rebalances // (typically achieves 60-80% reduction) + expect(simpleResult.kpis.totalRebalances).to.be.greaterThan( + 0, + 'SimpleRebalancer should have rebalanced at least once for ratio comparison', + ); const reductionRatio = productionResult.kpis.totalRebalances / simpleResult.kpis.totalRebalances; From a505c3e809c3e989383f5b885e4464dba86a543a Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 11:17:44 -0500 Subject: [PATCH 08/12] refactor(rebalancer-sim): Remove manual nonce management and use async loop Txs are sequential (await tx.wait()), so ethers handles nonces naturally. Replace setInterval + re-entry guard with simple while loop + sleep. Co-Authored-By: Claude Opus 4.6 --- .../src/MockInfrastructureController.ts | 151 ++++++++---------- 1 file changed, 63 insertions(+), 88 deletions(-) diff --git a/typescript/rebalancer-sim/src/MockInfrastructureController.ts b/typescript/rebalancer-sim/src/MockInfrastructureController.ts index 32da8c12e18..a91493fbef4 100644 --- a/typescript/rebalancer-sim/src/MockInfrastructureController.ts +++ b/typescript/rebalancer-sim/src/MockInfrastructureController.ts @@ -41,12 +41,8 @@ interface PendingMessage { */ export class MockInfrastructureController { private pendingMessages: PendingMessage[] = []; - private signer!: ethers.Signer; private isRunning = false; - private processing = false; - private processLoopTimer?: NodeJS.Timeout; - private currentNonce: number = 0; - private nonceInitialized = false; + private processLoopPromise?: Promise; constructor( private readonly core: HyperlaneCore, @@ -79,12 +75,6 @@ export class MockInfrastructureController { if (this.isRunning) return; this.isRunning = true; - // Get signer from MultiProvider for nonce management - const firstChain = this.core.multiProvider.getKnownChainNames()[0]; - this.signer = this.core.multiProvider.getSigner(firstChain); - this.currentNonce = await this.signer.getTransactionCount(); - this.nonceInitialized = true; - // Listen for Dispatch events on all mailboxes for (const chainName of this.core.multiProvider.getKnownChainNames()) { const mailbox = this.core.getContracts(chainName).mailbox; @@ -102,9 +92,7 @@ export class MockInfrastructureController { } // Start processing loop - this.processLoopTimer = setInterval(() => { - void this.processReadyMessages(); - }, 50); + this.processLoopPromise = this.processLoop(); } /** @@ -216,84 +204,70 @@ export class MockInfrastructureController { } /** - * Process messages that are past their delivery time. - * Retries indefinitely — waitForAllDeliveries handles the timeout. + * Async processing loop — delivers ready messages, sleeps between iterations. + * Retries indefinitely; waitForAllDeliveries handles the timeout. */ - private async processReadyMessages(): Promise { - if (this.processing) return; - this.processing = true; - try { - await this.doProcessReadyMessages(); - } finally { - this.processing = false; - } - } - - private async doProcessReadyMessages(): Promise { - const now = Date.now(); - const ready = this.pendingMessages.filter((m) => m.deliveryTime <= now); - if (ready.length === 0) return; + private async processLoop(): Promise { + while (this.isRunning) { + const now = Date.now(); + const ready = this.pendingMessages.filter((m) => m.deliveryTime <= now); + + for (const msg of ready) { + if (!this.isRunning) break; + + const mailbox = this.core.getContracts(msg.destination).mailbox; + + // Static call pre-check + try { + await mailbox.callStatic.process('0x', msg.message); + } catch (error) { + logger.debug( + { + messageId: msg.messageId, + dest: msg.destination, + attempts: msg.attempts, + error, + }, + 'Static pre-check failed, will retry', + ); + msg.attempts++; + msg.deliveryTime = now + 200; + continue; + } - if (!this.nonceInitialized) { - this.currentNonce = await this.signer.getTransactionCount(); - this.nonceInitialized = true; - } + try { + const tx = await mailbox.process('0x', msg.message); + await tx.wait(); - for (const msg of ready) { - const mailbox = this.core.getContracts(msg.destination).mailbox; - - // Static call pre-check - try { - await mailbox.callStatic.process('0x', msg.message); - } catch (error) { - logger.debug( - { - messageId: msg.messageId, - dest: msg.destination, - attempts: msg.attempts, - error, - }, - 'Static pre-check failed, will retry', - ); - msg.attempts++; - msg.deliveryTime = now + 200; - continue; - } + // Remove from pending + const idx = this.pendingMessages.indexOf(msg); + if (idx >= 0) this.pendingMessages.splice(idx, 1); - try { - const tx = await mailbox.process('0x', msg.message, { - nonce: this.currentNonce++, - }); - await tx.wait(); - - // Remove from pending - const idx = this.pendingMessages.indexOf(msg); - if (idx >= 0) this.pendingMessages.splice(idx, 1); - - // Record completion - if (msg.type === 'user-transfer') { - this.kpiCollector.recordTransferComplete(msg.messageId); - this.actionTracker?.removeTransfer(msg.messageId); - } else if (msg.type === 'bridge-transfer') { - this.kpiCollector.recordRebalanceComplete(msg.messageId); - if (this.actionTracker && msg.amount > 0n) { - this.actionTracker.completeRebalanceByRoute( - this.core.multiProvider.getDomainId(msg.origin), - this.core.multiProvider.getDomainId(msg.destination), - msg.amount, - ); + // Record completion + if (msg.type === 'user-transfer') { + this.kpiCollector.recordTransferComplete(msg.messageId); + this.actionTracker?.removeTransfer(msg.messageId); + } else if (msg.type === 'bridge-transfer') { + this.kpiCollector.recordRebalanceComplete(msg.messageId); + if (this.actionTracker && msg.amount > 0n) { + this.actionTracker.completeRebalanceByRoute( + this.core.multiProvider.getDomainId(msg.origin), + this.core.multiProvider.getDomainId(msg.destination), + msg.amount, + ); + } } + } catch (error) { + msg.attempts++; + msg.deliveryTime = now + 200; + logger.debug( + { messageId: msg.messageId, dest: msg.destination, error }, + 'Delivery tx failed, will retry', + ); } - } catch (error) { - msg.attempts++; - msg.deliveryTime = now + 200; - // Re-sync nonce on tx failure - this.currentNonce = await this.signer.getTransactionCount(); - logger.debug( - { messageId: msg.messageId, dest: msg.destination, error }, - 'Delivery tx failed, will retry', - ); } + + await new Promise((resolve) => setTimeout(resolve, 50)); } } @@ -303,9 +277,10 @@ export class MockInfrastructureController { async stop(): Promise { this.isRunning = false; - if (this.processLoopTimer) { - clearInterval(this.processLoopTimer); - this.processLoopTimer = undefined; + // Wait for the processing loop to exit + if (this.processLoopPromise) { + await this.processLoopPromise; + this.processLoopPromise = undefined; } for (const chainName of this.core.multiProvider.getKnownChainNames()) { From 9cbe67fa2c5e95b4f975054ff220c55993fef64e Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 11:27:53 -0500 Subject: [PATCH 09/12] fix(rebalancer-sim): Address PR review feedback round 3 - Import Transfer/RebalanceIntent/RebalanceAction from @hyperlane-xyz/rebalancer instead of redefining locally (export added to rebalancer index) - Catch unhandled rejections from async onDispatch in event listener - Guard domains[originChain] access in onDispatch Co-Authored-By: Claude Opus 4.6 --- .../src/MockInfrastructureController.ts | 16 +++++- .../src/runners/MockActionTracker.ts | 54 ++----------------- typescript/rebalancer/src/index.ts | 5 ++ 3 files changed, 24 insertions(+), 51 deletions(-) diff --git a/typescript/rebalancer-sim/src/MockInfrastructureController.ts b/typescript/rebalancer-sim/src/MockInfrastructureController.ts index a91493fbef4..2b3abf2edd6 100644 --- a/typescript/rebalancer-sim/src/MockInfrastructureController.ts +++ b/typescript/rebalancer-sim/src/MockInfrastructureController.ts @@ -86,7 +86,14 @@ export class MockInfrastructureController { _recipient: string, message: string, ) => { - void this.onDispatch(chainName, sender, destination, message); + this.onDispatch(chainName, sender, destination, message).catch( + (error: unknown) => { + logger.error( + { origin: chainName, error }, + 'Unhandled error in onDispatch', + ); + }, + ); }, ); } @@ -112,6 +119,13 @@ export class MockInfrastructureController { } const originDomain = this.domains[originChain]; + if (!originDomain) { + logger.warn( + { originChain }, + 'No domain config for origin chain, skipping', + ); + return; + } const senderLower = sender.toLowerCase(); // Classify by sender diff --git a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts index 1859ddfc77a..ade27a55ed6 100644 --- a/typescript/rebalancer-sim/src/runners/MockActionTracker.ts +++ b/typescript/rebalancer-sim/src/runners/MockActionTracker.ts @@ -2,61 +2,15 @@ import type { CreateRebalanceActionParams, CreateRebalanceIntentParams, IActionTracker, + RebalanceAction, + RebalanceIntent, + Transfer, } from '@hyperlane-xyz/rebalancer'; -import type { Address, Domain } from '@hyperlane-xyz/utils'; +import type { Domain } from '@hyperlane-xyz/utils'; import { rootLogger } from '@hyperlane-xyz/utils'; const logger = rootLogger.child({ module: 'MockActionTracker' }); -/** - * Transfer record matching the real Transfer type. - */ -interface Transfer { - id: string; - origin: Domain; - destination: Domain; - amount: bigint; - status: 'in_progress' | 'complete'; - messageId: string; - sender: Address; - recipient: Address; - createdAt: number; - updatedAt: number; -} - -/** - * RebalanceIntent record matching the real type. - */ -interface RebalanceIntent { - id: string; - origin: Domain; - destination: Domain; - amount: bigint; - bridge?: Address; - status: 'not_started' | 'in_progress' | 'complete' | 'cancelled' | 'failed'; - fulfilledAmount: bigint; - priority?: number; - strategyType?: string; - createdAt: number; - updatedAt: number; -} - -/** - * RebalanceAction record matching the real type. - */ -interface RebalanceAction { - id: string; - intentId: string; - origin: Domain; - destination: Domain; - amount: bigint; - messageId: string; - txHash?: string; - status: 'in_progress' | 'complete' | 'failed'; - createdAt: number; - updatedAt: number; -} - /** * Mock implementation of IActionTracker for simulation testing. * diff --git a/typescript/rebalancer/src/index.ts b/typescript/rebalancer/src/index.ts index 1c6bfaa95b8..0d2b2998523 100644 --- a/typescript/rebalancer/src/index.ts +++ b/typescript/rebalancer/src/index.ts @@ -98,6 +98,11 @@ export type { CreateRebalanceIntentParams, CreateRebalanceActionParams, } from './tracking/IActionTracker.js'; +export type { + Transfer, + RebalanceIntent, + RebalanceAction, +} from './tracking/types.js'; // Factory export { RebalancerContextFactory } from './factories/RebalancerContextFactory.js'; From b76c548a0a107cb1055ab37c15224ff2bfae4d50 Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 12:14:14 -0500 Subject: [PATCH 10/12] fix(cli): Enroll remote router on MockValueTransferBridge in e2e tests Bridge extends Router, so _Router_dispatch requires a remote router enrolled for the destination domain. Without it, transferRemote reverts and the SentTransferRemote event listener hangs forever. Co-Authored-By: Claude Opus 4.6 --- .../src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts index d8c771c7829..0586a4980fb 100644 --- a/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts +++ b/typescript/cli/src/tests/ethereum/warp/warp-rebalancer.e2e-test.ts @@ -977,6 +977,10 @@ describe('hyperlane warp rebalancer e2e tests', async function () { ethers.constants.AddressZero, ANVIL_DEPLOYER_ADDRESS, ); + await bridgeContract.enrollRemoteRouter( + destDomain, + addressToBytes32(destContractAddress), + ); // Allow bridge // This allow the bridge to be used to send the rebalance transaction @@ -1229,6 +1233,10 @@ describe('hyperlane warp rebalancer e2e tests', async function () { ethers.constants.AddressZero, ANVIL_DEPLOYER_ADDRESS, ); + await bridgeContract.enrollRemoteRouter( + destDomain, + addressToBytes32(destContractAddress), + ); // Allow bridge // This allow the bridge to be used to send the rebalance transaction From 29445951ddba270c9d4385532d3e690f921c190e Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 15:06:11 -0500 Subject: [PATCH 11/12] Add quoteing to mock value transfer bridge --- .../contracts/mock/MockValueTransferBridge.sol | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/solidity/contracts/mock/MockValueTransferBridge.sol b/solidity/contracts/mock/MockValueTransferBridge.sol index bc69d292295..6acd3b158d2 100644 --- a/solidity/contracts/mock/MockValueTransferBridge.sol +++ b/solidity/contracts/mock/MockValueTransferBridge.sol @@ -31,13 +31,18 @@ contract MockValueTransferBridge is Router, ITokenBridge { } function quoteTransferRemote( - uint32, //_destinationDomain, - bytes32, //_recipient, - uint256 //_amountOut + uint32 _destinationDomain, + bytes32 _recipient, + uint256 _amountOut ) public view virtual override returns (Quote[] memory) { - Quote[] memory quotes = new Quote[](1); - quotes[0] = Quote(collateral, 1); + uint256 dispatchFee = _Router_quoteDispatch( + _destinationDomain, + abi.encode(_recipient, _amountOut) + ); + Quote[] memory quotes = new Quote[](2); + quotes[0] = Quote(collateral, 1); + quotes[1] = Quote(address(0), dispatchFee); return quotes; } From 3d20bd78de02c31c6f520f49685c49fad2c8304a Mon Sep 17 00:00:00 2001 From: nambrot Date: Fri, 6 Feb 2026 15:54:26 -0500 Subject: [PATCH 12/12] fix: address PR review nits - Make collateral immutable in MockValueTransferBridge - Replace Record with Record - Log errors in catch blocks instead of swallowing silently - Extract magic numbers into named constants (MESSAGE_BODY_OFFSET, WARP_TOKEN_SCALE) Co-Authored-By: Claude Opus 4.6 --- .../contracts/mock/MockValueTransferBridge.sol | 2 +- .../src/MockInfrastructureController.ts | 18 ++++++++++-------- .../rebalancer-sim/src/SimulationEngine.ts | 11 ++++++----- 3 files changed, 17 insertions(+), 14 deletions(-) diff --git a/solidity/contracts/mock/MockValueTransferBridge.sol b/solidity/contracts/mock/MockValueTransferBridge.sol index 6acd3b158d2..ee4998acc42 100644 --- a/solidity/contracts/mock/MockValueTransferBridge.sol +++ b/solidity/contracts/mock/MockValueTransferBridge.sol @@ -9,7 +9,7 @@ import {ERC20Test} from "../test/ERC20Test.sol"; contract MockValueTransferBridge is Router, ITokenBridge { using SafeERC20 for IERC20; - address public collateral; + address public immutable collateral; event SentTransferRemote( uint32 indexed origin, diff --git a/typescript/rebalancer-sim/src/MockInfrastructureController.ts b/typescript/rebalancer-sim/src/MockInfrastructureController.ts index 2b3abf2edd6..541d8ad3101 100644 --- a/typescript/rebalancer-sim/src/MockInfrastructureController.ts +++ b/typescript/rebalancer-sim/src/MockInfrastructureController.ts @@ -14,6 +14,11 @@ import { DEFAULT_BRIDGE_ROUTE_CONFIG } from './types.js'; const logger = rootLogger.child({ module: 'MockInfrastructureController' }); +/** Hyperlane message body starts at byte offset 77 (version:1 + nonce:4 + origin:4 + sender:32 + dest:4 + recipient:32) */ +const MESSAGE_BODY_OFFSET = 77; +/** Warp tokens scale amounts by 10^decimals; simulation uses 18 decimals */ +const WARP_TOKEN_SCALE = BigInt(1e18); + /** Pending message awaiting delayed delivery */ interface PendingMessage { /** keccak256(message) — real Hyperlane messageId */ @@ -145,12 +150,7 @@ export class MockInfrastructureController { // Compute real messageId const messageId = ethers.utils.keccak256(message); - // Decode amount from body (offset 77 bytes into message) - // TokenRouter encodes: TokenMessage.format(recipient, _outboundAmount(amount)) - // where _outboundAmount = amount * scale. Scale = 10^decimals for warp tokens. - // We divide by scale to recover the original wei amount. - const bodyOffset = 77; - const body = '0x' + message.slice(2 + bodyOffset * 2); + const body = '0x' + message.slice(2 + MESSAGE_BODY_OFFSET * 2); let amount = 0n; try { const decoded = ethers.utils.defaultAbiCoder.decode( @@ -158,9 +158,11 @@ export class MockInfrastructureController { body, ); const scaledAmount = decoded[1].toBigInt(); - // Warp tokens use scale = 10^18, bridge Router uses scale = 1 (no scaling) + // Warp tokens use scale = 10^decimals, bridge Router uses scale = 1 (no scaling) amount = - type === 'user-transfer' ? scaledAmount / BigInt(1e18) : scaledAmount; + type === 'user-transfer' + ? scaledAmount / WARP_TOKEN_SCALE + : scaledAmount; } catch (error) { logger.warn( { messageId, origin: originChain, dest: destChain, error }, diff --git a/typescript/rebalancer-sim/src/SimulationEngine.ts b/typescript/rebalancer-sim/src/SimulationEngine.ts index 24ce151e7a6..dcdae674422 100644 --- a/typescript/rebalancer-sim/src/SimulationEngine.ts +++ b/typescript/rebalancer-sim/src/SimulationEngine.ts @@ -5,6 +5,7 @@ import { HypERC20Collateral__factory, } from '@hyperlane-xyz/core'; import { + type ChainMetadata, HyperlaneCore, MultiProvider, TokenStandard, @@ -136,15 +137,15 @@ export class SimulationEngine { try { await rebalancer.stop(); - } catch { - // Ignore stop errors + } catch (error: unknown) { + logger.debug({ error }, 'Rebalancer stop failed during cleanup'); } if (controller) { try { await controller.stop(); - } catch { - // Ignore stop errors + } catch (error: unknown) { + logger.debug({ error }, 'Controller stop failed during cleanup'); } } @@ -276,7 +277,7 @@ export class SimulationEngine { * mailbox processor signer set on all chains. */ private buildHyperlaneCore(): HyperlaneCore { - const chainMetadata: Record = {}; + const chainMetadata: Record = {}; const addressesMap: Record = {}; for (const [chainName, domain] of Object.entries(this.deployment.domains)) { chainMetadata[chainName] = {