1ec308c3a0
Some checks failed
CI/CD Pipeline / Solidity Contracts (push) Failing after 1m12s
CI/CD Pipeline / Security Scanning (push) Successful in 2m21s
CI/CD Pipeline / Lint and Format (push) Failing after 25s
CI/CD Pipeline / Terraform Validation (push) Failing after 24s
CI/CD Pipeline / Kubernetes Validation (push) Successful in 22s
Deploy ChainID 138 / Deploy ChainID 138 (push) Failing after 41s
Validation / validate-genesis (push) Successful in 27s
Validation / validate-terraform (push) Failing after 25s
Validation / validate-kubernetes (push) Failing after 9s
Validation / validate-smart-contracts (push) Failing after 9s
Validation / validate-security (push) Failing after 1m6s
Validation / validate-documentation (push) Failing after 16s
OMNL reconcile anchor / Run omnl:reconcile and upload artifacts (push) Failing after 23s
Verify Deployment / Verify Deployment (push) Failing after 13m52s
Includes Monad↔Chain138 CCIP proof/deploy/verify tooling, relay service guards, bridge integration tweaks, and frontend ENS/network config follow-ups. Co-authored-by: Cursor <cursoragent@cursor.com>
1220 lines
44 KiB
JavaScript
1220 lines
44 KiB
JavaScript
/**
|
|
* CCIP Relay Service
|
|
* Monitors MessageSent events and relays messages to destination chain
|
|
*/
|
|
|
|
import { ethers, NonceManager } from 'ethers';
|
|
import { MessageSentABI, RelayRouterABI, RelayBridgeABI, ERC20ABI } from './abis.js';
|
|
import { MessageQueue } from './MessageQueue.js';
|
|
import { RpcUrlPool } from './rpcPool.js';
|
|
import {
|
|
isRelayShedding,
|
|
getRelaySheddingSourcePollIntervalMs,
|
|
getRelaySheddingQueueIdleMs
|
|
} from './config.js';
|
|
|
|
export class RelayService {
|
|
constructor(config, logger) {
|
|
this.config = config;
|
|
this.logger = logger;
|
|
this.isRunning = false;
|
|
this.sourceProvider = null;
|
|
this.destinationProvider = null;
|
|
this.destinationSubmitProvider = null;
|
|
this.destinationRpcPool = null;
|
|
this._readProviders = new Map();
|
|
this._activeRelayCount = 0;
|
|
this.sourceSigner = null;
|
|
this.destinationSigner = null;
|
|
this.messageQueue = new MessageQueue(this.logger, {
|
|
persistencePath: config.queuePersistence?.path || ''
|
|
});
|
|
|
|
// Contract instances
|
|
this.sourceRouter = null;
|
|
this.destinationRelayRouter = null;
|
|
this.destinationRelayBridge = null;
|
|
this.destinationBridgeContracts = new Map();
|
|
this.messageSentInterface = new ethers.Interface(MessageSentABI);
|
|
/** @type {number} throttle for shedding warning logs */
|
|
this._lastSheddingLogTs = 0;
|
|
this.startedAt = new Date().toISOString();
|
|
this.startEpochMs = Date.now();
|
|
this.relayerAddress = '';
|
|
this.lastSourcePoll = null;
|
|
this.lastSeenMessage = null;
|
|
this.lastRelayAttempt = null;
|
|
this.lastRelaySuccess = null;
|
|
this.lastError = null;
|
|
}
|
|
|
|
getReadProvider(url) {
|
|
const key = String(url || '');
|
|
if (!this._readProviders.has(key)) {
|
|
this._readProviders.set(key, new ethers.JsonRpcProvider(key));
|
|
}
|
|
return this._readProviders.get(key);
|
|
}
|
|
|
|
getMaxConcurrent() {
|
|
const configured = this.config.concurrency?.maxConcurrent;
|
|
if (Number.isFinite(configured) && configured >= 1) {
|
|
return Math.min(Math.floor(configured), 12);
|
|
}
|
|
const fromEnv = parseInt(process.env.RELAY_MAX_CONCURRENT || '1', 10);
|
|
if (!Number.isFinite(fromEnv) || fromEnv < 1) return 1;
|
|
return Math.min(fromEnv, 12);
|
|
}
|
|
|
|
async probeBridgeProcessed(messageId, targetBridge, url) {
|
|
const bridge = new ethers.Contract(
|
|
targetBridge,
|
|
RelayBridgeABI,
|
|
this.getReadProvider(url)
|
|
);
|
|
if (bridge.processed) {
|
|
if (await bridge.processed(messageId)) return true;
|
|
}
|
|
if (bridge.processedTransfers) {
|
|
if (await bridge.processedTransfers(messageId)) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
async getDestinationBridgeContract(targetBridge) {
|
|
const key = String(targetBridge).toLowerCase();
|
|
let contract = this.destinationBridgeContracts.get(key);
|
|
if (!contract) {
|
|
contract = new ethers.Contract(targetBridge, RelayBridgeABI, this.destinationProvider);
|
|
this.destinationBridgeContracts.set(key, contract);
|
|
}
|
|
return contract;
|
|
}
|
|
|
|
/** True when destination bridge already recorded this messageId (no relay tx needed). */
|
|
async isDeliveredOnDestination(messageId, targetBridge) {
|
|
if (process.env.RELAY_SKIP_DESTINATION_PROCESSED_PROBE === '1') {
|
|
return false;
|
|
}
|
|
if (!targetBridge) {
|
|
return false;
|
|
}
|
|
try {
|
|
if (this.destinationRpcPool && this.destinationRpcPool.size > 0) {
|
|
return await this.destinationRpcPool.withFailover(
|
|
'processed probe',
|
|
(url) => this.probeBridgeProcessed(messageId, targetBridge, url)
|
|
);
|
|
}
|
|
const bridge = await this.getDestinationBridgeContract(targetBridge);
|
|
if (bridge.processed) {
|
|
if (await bridge.processed(messageId)) {
|
|
return true;
|
|
}
|
|
}
|
|
if (bridge.processedTransfers) {
|
|
if (await bridge.processedTransfers(messageId)) {
|
|
return true;
|
|
}
|
|
}
|
|
} catch (probeErr) {
|
|
this.logger.debug(`Destination processed probe failed for ${messageId}`, probeErr);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
normalizeAddress(value) {
|
|
if (!value) return '';
|
|
try {
|
|
return ethers.getAddress(value);
|
|
} catch (_) {
|
|
return '';
|
|
}
|
|
}
|
|
|
|
getConfiguredSourceBridge() {
|
|
return this.normalizeAddress(this.config.sourceChain.bridgeAddress);
|
|
}
|
|
|
|
getDestinationBridgeAllowlist() {
|
|
return (this.config.destinationChain.relayBridgeAllowlist || [])
|
|
.map((value) => String(value || '').trim().toLowerCase())
|
|
.filter(Boolean);
|
|
}
|
|
|
|
resolveTargetBridge(receiver) {
|
|
let targetBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeAddress);
|
|
try {
|
|
if (receiver) {
|
|
const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address'], receiver);
|
|
if (decoded && decoded[0]) {
|
|
targetBridge = this.normalizeAddress(decoded[0]) || targetBridge;
|
|
}
|
|
}
|
|
} catch (_) {
|
|
// Keep the configured fallback bridge when receiver decode fails.
|
|
}
|
|
return targetBridge;
|
|
}
|
|
|
|
resolveBridgeForMessage(messageData) {
|
|
const linkBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeLinkAddress);
|
|
const wethBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeAddress);
|
|
const tokenAmounts = messageData.tokenAmounts || [];
|
|
if (tokenAmounts.length > 0 && linkBridge) {
|
|
const sourceToken = String(tokenAmounts[0].token || '').toLowerCase();
|
|
const link138 = '0xb7721dd53a8c629d9f1ba31a5819afe250002b03';
|
|
if (sourceToken === link138) {
|
|
return linkBridge;
|
|
}
|
|
}
|
|
return this.resolveTargetBridge(messageData.receiver) || wethBridge;
|
|
}
|
|
|
|
mapSourceTokenToDestination(sourceToken) {
|
|
const normalized = this.normalizeAddress(sourceToken);
|
|
if (!normalized) return sourceToken;
|
|
const mapping = this.config.tokenMapping || {};
|
|
return (
|
|
mapping[normalized] ||
|
|
mapping[normalized.toLowerCase()] ||
|
|
normalized
|
|
);
|
|
}
|
|
|
|
normalizeRelayData(data, receiver, tokenAmounts, targetBridge, sender) {
|
|
let normalizedData = data;
|
|
const emptyData = !data || data === '0x' || data === '0x0';
|
|
const linkBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeLinkAddress);
|
|
const target = this.normalizeAddress(targetBridge);
|
|
|
|
if (emptyData && linkBridge && target && target.toLowerCase() === linkBridge.toLowerCase()) {
|
|
let recipient = '';
|
|
try {
|
|
if (receiver) {
|
|
const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address'], receiver);
|
|
recipient = this.normalizeAddress(decoded[0]);
|
|
}
|
|
} catch (_) {
|
|
// Fall through to optional sender-based recipient below.
|
|
}
|
|
if (!recipient && sender) {
|
|
recipient = this.normalizeAddress(sender);
|
|
}
|
|
const amount = tokenAmounts.length > 0
|
|
? (typeof tokenAmounts[0].amount === 'bigint'
|
|
? tokenAmounts[0].amount
|
|
: BigInt(tokenAmounts[0].amount.toString()))
|
|
: 0n;
|
|
if (recipient && amount > 0n) {
|
|
normalizedData = ethers.AbiCoder.defaultAbiCoder().encode(
|
|
['address', 'uint256'],
|
|
[recipient, amount]
|
|
);
|
|
this.logger.info(
|
|
`Synthesized LINK bridge payload for ${recipient} amount ${amount.toString()}`
|
|
);
|
|
}
|
|
}
|
|
|
|
if (process.env.RELAY_EXPAND_TWO_FIELD_PAYLOAD === '1') {
|
|
try {
|
|
const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address', 'uint256'], normalizedData);
|
|
const reencoded2 = ethers.AbiCoder.defaultAbiCoder().encode(['address', 'uint256'], [decoded[0], decoded[1]]);
|
|
if ((normalizedData || '').toLowerCase() === reencoded2.toLowerCase()) {
|
|
normalizedData = ethers.AbiCoder.defaultAbiCoder().encode(
|
|
['address', 'uint256', 'address', 'uint256'],
|
|
[decoded[0], decoded[1], sender, 0]
|
|
);
|
|
this.logger.debug('Normalized 2-field payload to 4-field payload');
|
|
}
|
|
} catch (_) {
|
|
// Keep original payload when it does not match (address,uint256).
|
|
}
|
|
}
|
|
|
|
return normalizedData;
|
|
}
|
|
|
|
evaluateMessageScope(messageData) {
|
|
const sourceBridge = this.getConfiguredSourceBridge();
|
|
const sender = this.normalizeAddress(messageData.sender);
|
|
const targetBridge = this.resolveBridgeForMessage(messageData);
|
|
const allowlist = this.getDestinationBridgeAllowlist();
|
|
|
|
if (sourceBridge && sender && sender.toLowerCase() !== sourceBridge.toLowerCase()) {
|
|
return {
|
|
inScope: false,
|
|
reason: `sender ${sender} does not match worker source bridge ${sourceBridge}`,
|
|
sourceBridge,
|
|
sender,
|
|
targetBridge
|
|
};
|
|
}
|
|
|
|
if (!targetBridge) {
|
|
return {
|
|
inScope: false,
|
|
reason: 'destination bridge could not be resolved from receiver or config',
|
|
sourceBridge,
|
|
sender,
|
|
targetBridge: ''
|
|
};
|
|
}
|
|
|
|
if (allowlist.length > 0 && !allowlist.includes(targetBridge.toLowerCase())) {
|
|
return {
|
|
inScope: false,
|
|
reason: `destination bridge ${targetBridge} not in worker allowlist`,
|
|
sourceBridge,
|
|
sender,
|
|
targetBridge
|
|
};
|
|
}
|
|
|
|
return {
|
|
inScope: true,
|
|
reason: '',
|
|
sourceBridge,
|
|
sender,
|
|
targetBridge
|
|
};
|
|
}
|
|
|
|
static summarizeError(error) {
|
|
if (!error) return '';
|
|
if (typeof error === 'string') return error;
|
|
return String(error.shortMessage || error.message || error);
|
|
}
|
|
|
|
/**
|
|
* Fail fast when RPC URL points at the wrong chain (e.g. Avalanche RPC + mainnet DEST_CHAIN_ID).
|
|
*/
|
|
async assertNetworkAlignment(label, provider, expectedChainId) {
|
|
const network = await provider.getNetwork();
|
|
const actual = Number(network.chainId);
|
|
const expected = Number(expectedChainId);
|
|
if (!Number.isFinite(expected) || expected <= 0) {
|
|
throw new Error(`${label} expected chainId not configured (${expectedChainId})`);
|
|
}
|
|
if (actual !== expected) {
|
|
throw new Error(
|
|
`${label} RPC chain-id mismatch: provider reports ${actual} but config expects ${expected}. ` +
|
|
'Check DEST_RPC_URL / RPC_URL_MAINNET and .env.local (named profiles skip .env.local by default).'
|
|
);
|
|
}
|
|
this.logger.info('%s RPC chain-id OK (%s)', label, actual);
|
|
}
|
|
|
|
recordError(scope, error, extra = {}) {
|
|
this.lastError = {
|
|
at: new Date().toISOString(),
|
|
scope,
|
|
message: RelayService.summarizeError(error),
|
|
...extra
|
|
};
|
|
}
|
|
|
|
getHealthStatus() {
|
|
if (!this.isRunning) {
|
|
return 'stopped';
|
|
}
|
|
if (isRelayShedding()) {
|
|
return 'paused';
|
|
}
|
|
|
|
if (this.lastSourcePoll && this.lastSourcePoll.ok === false) {
|
|
return 'degraded';
|
|
}
|
|
|
|
if (!this.lastSourcePoll || !this.lastSourcePoll.at) {
|
|
return 'starting';
|
|
}
|
|
|
|
const lastPollMs = Date.parse(this.lastSourcePoll.at);
|
|
const staleAfterMs = Math.max(this.getSourcePollIntervalMs() * 3, 15000);
|
|
if (Number.isFinite(lastPollMs) && Date.now() - lastPollMs > staleAfterMs) {
|
|
return 'stale';
|
|
}
|
|
|
|
const lastErrorMs = this.lastError && this.lastError.at ? Date.parse(this.lastError.at) : 0;
|
|
const lastSuccessMs = this.lastRelaySuccess && this.lastRelaySuccess.at ? Date.parse(this.lastRelaySuccess.at) : 0;
|
|
if (
|
|
this.lastError &&
|
|
(
|
|
this.lastError.scope === 'relay_message' ||
|
|
this.lastError.scope === 'bridge_inventory' ||
|
|
this.lastError.scope === 'bridge_inventory_probe'
|
|
) &&
|
|
Number.isFinite(lastErrorMs) &&
|
|
lastErrorMs > 0 &&
|
|
lastErrorMs >= lastSuccessMs
|
|
) {
|
|
return 'degraded';
|
|
}
|
|
|
|
return 'operational';
|
|
}
|
|
|
|
async ensureTargetBridgeInventory(messageId, targetBridge, tokenAmounts) {
|
|
if (process.env.RELAY_ENFORCE_BRIDGE_TOKEN_BALANCE !== '1') {
|
|
return { ok: true };
|
|
}
|
|
|
|
for (const tokenAmount of tokenAmounts) {
|
|
const tokenAddress = ethers.getAddress(tokenAmount.token);
|
|
const requiredAmount = typeof tokenAmount.amount === 'bigint'
|
|
? tokenAmount.amount
|
|
: BigInt(tokenAmount.amount.toString());
|
|
|
|
const tokenContract = new ethers.Contract(tokenAddress, ERC20ABI, this.destinationProvider);
|
|
const availableAmount = await tokenContract.balanceOf(targetBridge);
|
|
if (availableAmount < requiredAmount) {
|
|
const shortfall = requiredAmount - availableAmount;
|
|
return {
|
|
ok: false,
|
|
token: tokenAddress,
|
|
requiredAmount,
|
|
availableAmount,
|
|
shortfall,
|
|
message: `Insufficient bridge inventory for ${messageId}: ${tokenAddress} available=${availableAmount.toString()} required=${requiredAmount.toString()} shortfall=${shortfall.toString()}`
|
|
};
|
|
}
|
|
}
|
|
|
|
return { ok: true };
|
|
}
|
|
|
|
getHealthSnapshot() {
|
|
const queueStats = this.messageQueue.getStats();
|
|
const status = this.getHealthStatus();
|
|
const sourceBridge = this.getConfiguredSourceBridge();
|
|
const allowlist = this.getDestinationBridgeAllowlist();
|
|
const defaultTargetBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeAddress);
|
|
|
|
return {
|
|
ok: status === 'operational' || status === 'paused' || status === 'starting',
|
|
status,
|
|
service: {
|
|
name: 'ccip-relay',
|
|
running: this.isRunning,
|
|
pid: process.pid,
|
|
profile: process.env.RELAY_PROFILE || 'default',
|
|
started_at: this.startedAt,
|
|
uptime_sec: Math.max(0, Math.floor((Date.now() - this.startEpochMs) / 1000)),
|
|
relayer_address: this.relayerAddress || this.config.relayer.address || ''
|
|
},
|
|
source: {
|
|
chain_name: this.config.sourceChain.name,
|
|
chain_id: this.config.sourceChain.chainId,
|
|
chain_selector: this.config.sourceChainSelector.toString(),
|
|
router_address: this.config.sourceChain.routerAddress,
|
|
bridge_address: this.config.sourceChain.bridgeAddress,
|
|
bridge_filter: sourceBridge || ''
|
|
},
|
|
destination: {
|
|
chain_name: this.config.destinationChain.name,
|
|
chain_id: this.config.destinationChain.chainId,
|
|
chain_selector: this.config.destinationChain.chainSelector.toString(),
|
|
relay_router: this.config.destinationChain.relayRouterAddress,
|
|
relay_bridge: this.config.destinationChain.relayBridgeAddress,
|
|
relay_bridge_default: defaultTargetBridge || '',
|
|
relay_bridge_allowlist: allowlist,
|
|
delivery_mode: this.config.destinationChain.deliveryMode
|
|
},
|
|
scope: {
|
|
source_bridge: sourceBridge || '',
|
|
destination_bridge_default: defaultTargetBridge || '',
|
|
destination_bridge_allowlist: allowlist
|
|
},
|
|
monitoring: {
|
|
start_block: String(this.config.monitoring.startBlock),
|
|
poll_interval_ms: this.config.monitoring.pollInterval,
|
|
effective_source_poll_interval_ms: this.getSourcePollIntervalMs(),
|
|
confirmation_blocks: this.config.monitoring.confirmationBlocks,
|
|
finality_delay_blocks: this.config.monitoring.finalityDelayBlocks || 0,
|
|
replay_window_blocks: this.config.monitoring.replayWindowBlocks || 0,
|
|
shedding: isRelayShedding(),
|
|
delivery_enabled: !isRelayShedding()
|
|
},
|
|
queue: {
|
|
size: queueStats.queueSize,
|
|
processed: queueStats.processed,
|
|
failed: queueStats.failed,
|
|
in_flight: queueStats.inFlight,
|
|
persistence_enabled: queueStats.persistenceEnabled,
|
|
last_persisted_at: queueStats.lastPersistedAt
|
|
},
|
|
concurrency: {
|
|
max_concurrent: this.getMaxConcurrent(),
|
|
active_relay_tasks: this._activeRelayCount
|
|
},
|
|
last_source_poll: this.lastSourcePoll,
|
|
last_seen_message: this.lastSeenMessage,
|
|
last_relay_attempt: this.lastRelayAttempt,
|
|
last_relay_success: this.lastRelaySuccess,
|
|
last_error: this.lastError
|
|
};
|
|
}
|
|
|
|
shouldSkipMessageId(messageId) {
|
|
const normalized = String(messageId || '').toLowerCase();
|
|
return normalized && this.config.skipMessageIds && this.config.skipMessageIds.has(normalized);
|
|
}
|
|
|
|
/** Polling interval for source router logs (longer while shedding to cut RPC churn). */
|
|
getSourcePollIntervalMs() {
|
|
if (isRelayShedding()) {
|
|
return getRelaySheddingSourcePollIntervalMs();
|
|
}
|
|
return this.config.monitoring.pollInterval;
|
|
}
|
|
|
|
async getDestinationTxOptions() {
|
|
const txOptions = {
|
|
gasLimit: BigInt(process.env.RELAY_DEST_GAS_LIMIT || '1000000')
|
|
};
|
|
|
|
if (process.env.RELAY_DEST_LEGACY_TX !== '1') {
|
|
return txOptions;
|
|
}
|
|
|
|
let gasPrice = process.env.RELAY_DEST_GAS_PRICE_WEI ? BigInt(process.env.RELAY_DEST_GAS_PRICE_WEI) : null;
|
|
if (gasPrice === null) {
|
|
try {
|
|
const feeData = await this.destinationProvider.getFeeData();
|
|
if (feeData.gasPrice) {
|
|
gasPrice = feeData.gasPrice;
|
|
}
|
|
} catch (_) {
|
|
// Fall through to raw RPC fallback below.
|
|
}
|
|
}
|
|
|
|
if (gasPrice === null) {
|
|
const gasProvider = this.destinationSubmitProvider || this.destinationProvider;
|
|
const rawGasPrice = await gasProvider.send('eth_gasPrice', []);
|
|
gasPrice = BigInt(rawGasPrice);
|
|
}
|
|
|
|
const bumpPct = BigInt(process.env.RELAY_DEST_GAS_PRICE_BUMP_PCT || '20');
|
|
const bumpWei = BigInt(process.env.RELAY_DEST_GAS_PRICE_BUMP_WEI || '1000000');
|
|
gasPrice = gasPrice + (gasPrice * bumpPct) / 100n + bumpWei;
|
|
|
|
txOptions.type = 0;
|
|
txOptions.gasPrice = gasPrice;
|
|
return txOptions;
|
|
}
|
|
|
|
/** Prevent indefinite hang when mainnet confirmation stalls (blocks queue processor). */
|
|
async waitForDestinationReceipt(tx, messageId) {
|
|
const timeoutMs = parseInt(process.env.RELAY_TX_CONFIRM_TIMEOUT_MS || '180000', 10);
|
|
const safeTimeout = Number.isFinite(timeoutMs) && timeoutMs >= 30_000 ? timeoutMs : 180_000;
|
|
let timer;
|
|
try {
|
|
return await Promise.race([
|
|
tx.wait(),
|
|
new Promise((_, reject) => {
|
|
timer = setTimeout(
|
|
() =>
|
|
reject(
|
|
new Error(
|
|
`tx.wait timed out after ${safeTimeout}ms for ${messageId}${tx?.hash ? ` (${tx.hash})` : ''}`
|
|
)
|
|
),
|
|
safeTimeout
|
|
);
|
|
})
|
|
]);
|
|
} finally {
|
|
if (timer) clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
async start() {
|
|
this.logger.info('Initializing relay service...');
|
|
|
|
const poolUrls =
|
|
(this.config.destinationChain.rpcPoolUrls && this.config.destinationChain.rpcPoolUrls.length > 0)
|
|
? this.config.destinationChain.rpcPoolUrls
|
|
: [this.config.destinationChain.rpcUrl];
|
|
this.destinationRpcPool = new RpcUrlPool(poolUrls, this.logger);
|
|
|
|
const submitUrl = this.config.destinationChain.submitRpcUrl || poolUrls[0];
|
|
|
|
this.sourceProvider = new ethers.JsonRpcProvider(this.config.sourceChain.rpcUrl);
|
|
this.destinationProvider = new ethers.JsonRpcProvider(poolUrls[0]);
|
|
this.destinationSubmitProvider =
|
|
submitUrl === poolUrls[0]
|
|
? this.destinationProvider
|
|
: new ethers.JsonRpcProvider(submitUrl);
|
|
|
|
await this.assertNetworkAlignment('source', this.sourceProvider, this.config.sourceChain.chainId);
|
|
await this.assertNetworkAlignment('destination', this.destinationProvider, this.config.destinationChain.chainId);
|
|
if (this.destinationSubmitProvider !== this.destinationProvider) {
|
|
await this.assertNetworkAlignment(
|
|
'destination-submit',
|
|
this.destinationSubmitProvider,
|
|
this.config.destinationChain.chainId
|
|
);
|
|
this.logger.info('Destination submit RPC: %s', submitUrl);
|
|
}
|
|
|
|
if (!this.config.relayer.privateKey) {
|
|
throw new Error('Relayer private key not configured');
|
|
}
|
|
this.sourceSigner = new ethers.Wallet(this.config.relayer.privateKey, this.sourceProvider);
|
|
const destinationWallet = new ethers.Wallet(
|
|
this.config.relayer.privateKey,
|
|
this.destinationSubmitProvider
|
|
);
|
|
this.destinationSigner = new NonceManager(destinationWallet);
|
|
this.relayerAddress = String(destinationWallet.address);
|
|
await this.messageQueue.init();
|
|
|
|
this.logger.info('Relayer address: %s', this.relayerAddress);
|
|
this.logger.info(
|
|
'Relay concurrency: maxConcurrent=%s rpcPoolSize=%s',
|
|
this.getMaxConcurrent(),
|
|
poolUrls.length
|
|
);
|
|
|
|
// Validate relay router address (bridge can be dynamic from message receiver)
|
|
if (!this.config.destinationChain.relayRouterAddress ||
|
|
this.config.destinationChain.relayRouterAddress === '') {
|
|
throw new Error(`Relay router address must be configured on destination chain. Router: ${this.config.destinationChain.relayRouterAddress}`);
|
|
}
|
|
|
|
// Initialize contract instances
|
|
this.sourceRouter = new ethers.Contract(
|
|
this.config.sourceChain.routerAddress,
|
|
MessageSentABI,
|
|
this.sourceProvider
|
|
);
|
|
|
|
this.destinationRelayRouter = new ethers.Contract(
|
|
this.config.destinationChain.relayRouterAddress,
|
|
RelayRouterABI,
|
|
this.destinationSigner
|
|
);
|
|
|
|
// Backward-compatible static destination bridge (optional)
|
|
if (this.config.destinationChain.relayBridgeAddress) {
|
|
this.destinationRelayBridge = new ethers.Contract(
|
|
this.config.destinationChain.relayBridgeAddress,
|
|
RelayBridgeABI,
|
|
this.destinationProvider
|
|
);
|
|
}
|
|
|
|
// Start monitoring and queue drain concurrently (monitoring loop never returns).
|
|
this.isRunning = true;
|
|
void this.startProcessingQueue();
|
|
await this.startMonitoring();
|
|
}
|
|
|
|
async stop() {
|
|
this.logger.info('Stopping relay service...');
|
|
this.isRunning = false;
|
|
const deadline = Date.now() + 30_000;
|
|
while (this._activeRelayCount > 0 && Date.now() < deadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 500));
|
|
}
|
|
await this.messageQueue.persistSnapshot();
|
|
}
|
|
|
|
/** Preferred chunk size for scanning (adaptive split handles stricter RPCs). Override: SOURCE_LOGS_MAX_BLOCK_RANGE */
|
|
getSourceLogsMaxBlockRange() {
|
|
const v = parseInt(
|
|
process.env.SOURCE_LOGS_MAX_BLOCK_RANGE || process.env.RELAY_SOURCE_LOGS_MAX_BLOCK_RANGE || '8000',
|
|
10
|
|
);
|
|
return Number.isFinite(v) && v >= 500 ? v : 8000;
|
|
}
|
|
|
|
static _isRpcLogRangeError(err) {
|
|
const msg = String(err && err.message ? err.message : err);
|
|
return (
|
|
msg.includes('maximum RPC range') ||
|
|
msg.includes('exceeds maximum') ||
|
|
msg.includes('requested too many blocks') ||
|
|
msg.includes('maximum is set to') ||
|
|
(msg.includes('-32000') && (
|
|
msg.includes('range') ||
|
|
msg.includes('too many blocks') ||
|
|
msg.includes('maximum is set to')
|
|
))
|
|
);
|
|
}
|
|
|
|
async fetchSourceLogs(fromBlock, toBlock) {
|
|
const payload = {
|
|
jsonrpc: '2.0',
|
|
id: 1,
|
|
method: 'eth_getLogs',
|
|
params: [{
|
|
address: this.config.sourceChain.routerAddress,
|
|
fromBlock: ethers.toQuantity(fromBlock),
|
|
toBlock: ethers.toQuantity(toBlock),
|
|
topics: [this.messageSentInterface.getEvent('MessageSent').topicHash]
|
|
}]
|
|
};
|
|
|
|
const response = await fetch(this.config.sourceChain.rpcUrl, {
|
|
method: 'POST',
|
|
headers: { 'content-type': 'application/json' },
|
|
body: JSON.stringify(payload)
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new Error(`eth_getLogs HTTP ${response.status}`);
|
|
}
|
|
|
|
const data = await response.json();
|
|
if (data.error) {
|
|
throw new Error(`eth_getLogs RPC ${data.error.code}: ${data.error.message}`);
|
|
}
|
|
|
|
return Array.isArray(data.result) ? data.result : [];
|
|
}
|
|
|
|
_sortRawLogs(all) {
|
|
all.sort((a, b) => {
|
|
const ba = Number(BigInt(a.blockNumber));
|
|
const bb = Number(BigInt(b.blockNumber));
|
|
if (ba !== bb) return ba - bb;
|
|
return String(a.logIndex || '').localeCompare(String(b.logIndex || ''));
|
|
});
|
|
return all;
|
|
}
|
|
|
|
/**
|
|
* Single eth_getLogs for [fromBlock,toBlock]; on RPC range-limit errors, bisect until requests fit.
|
|
*/
|
|
async fetchSourceLogsAdaptive(fromBlock, toBlock) {
|
|
if (toBlock < fromBlock) return [];
|
|
try {
|
|
return await this.fetchSourceLogs(fromBlock, toBlock);
|
|
} catch (e) {
|
|
if (!RelayService._isRpcLogRangeError(e) || fromBlock === toBlock) throw e;
|
|
const mid = Math.floor((fromBlock + toBlock) / 2);
|
|
const left = await this.fetchSourceLogsAdaptive(fromBlock, mid);
|
|
const right = await this.fetchSourceLogsAdaptive(mid + 1, toBlock);
|
|
return this._sortRawLogs([...left, ...right]);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Scan MessageSent logs from fromBlock..toBlock using coarse chunks, each resolved via adaptive splits.
|
|
*/
|
|
async fetchSourceLogsChunked(fromBlock, toBlock) {
|
|
const maxSpan = this.getSourceLogsMaxBlockRange();
|
|
if (toBlock < fromBlock) return [];
|
|
|
|
const all = [];
|
|
let cursor = fromBlock;
|
|
while (cursor <= toBlock && this.isRunning) {
|
|
const chunkEnd = Math.min(toBlock, cursor + maxSpan - 1);
|
|
const chunk = await this.fetchSourceLogsAdaptive(cursor, chunkEnd);
|
|
if (chunk.length) all.push(...chunk);
|
|
cursor = chunkEnd + 1;
|
|
}
|
|
return this._sortRawLogs(all);
|
|
}
|
|
|
|
async startMonitoring() {
|
|
this.logger.info('Starting event monitoring...');
|
|
|
|
let startBlock;
|
|
if (this.config.monitoring.startBlock === 'latest' || isNaN(this.config.monitoring.startBlock)) {
|
|
const currentBlock = await this.sourceProvider.getBlockNumber();
|
|
startBlock = currentBlock > 0 ? currentBlock - 1 : 0;
|
|
} else {
|
|
startBlock = parseInt(this.config.monitoring.startBlock);
|
|
}
|
|
|
|
this.logger.info(`Monitoring from block: ${startBlock}`);
|
|
|
|
// HTTP filter subscriptions are unreliable on several RPCs used here.
|
|
// Polling via queryFilter is the default and can be overridden explicitly.
|
|
if (process.env.RELAY_USE_EVENT_SUBSCRIPTIONS === '1') {
|
|
this.sourceRouter.on('MessageSent', async (messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs, event) => {
|
|
try {
|
|
const destSelector = destinationChainSelector.toString();
|
|
const expectedSelector = this.config.destinationChain.chainSelector.toString();
|
|
|
|
if (this.shouldSkipMessageId(messageId)) {
|
|
this.logger.warn(`Skipping MessageSent ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
|
|
return;
|
|
}
|
|
|
|
if (destSelector !== expectedSelector) {
|
|
this.logger.debug(`Ignoring message for different chain: ${destSelector}`);
|
|
return;
|
|
}
|
|
|
|
const scope = this.evaluateMessageScope({ messageId, sender, receiver, tokenAmounts });
|
|
if (!scope.inScope) {
|
|
this.logger.debug(`Ignoring message ${messageId}; ${scope.reason}`);
|
|
return;
|
|
}
|
|
|
|
this.logger.info('MessageSent event detected:', {
|
|
messageId: messageId,
|
|
destinationChainSelector: destinationChainSelector.toString(),
|
|
sender: sender,
|
|
targetBridge: scope.targetBridge,
|
|
blockNumber: event.blockNumber,
|
|
transactionHash: event.transactionHash
|
|
});
|
|
|
|
const receipt = await event.getTransactionReceipt();
|
|
const confirmations = this.config.monitoring.confirmationBlocks;
|
|
|
|
if (confirmations > 0) {
|
|
await receipt.confirmations(confirmations);
|
|
}
|
|
|
|
const formattedTokenAmounts = tokenAmounts.map(ta => ({
|
|
token: ta.token,
|
|
amount: ta.amount,
|
|
amountType: ta.amountType
|
|
}));
|
|
|
|
if (await this.isDeliveredOnDestination(messageId, scope.targetBridge)) {
|
|
this.logger.info(`Message ${messageId} already on destination; recording processed (live event)`);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return;
|
|
}
|
|
|
|
await this.messageQueue.add({
|
|
messageId,
|
|
destinationChainSelector,
|
|
sender,
|
|
receiver,
|
|
targetBridge: scope.targetBridge,
|
|
data,
|
|
tokenAmounts: formattedTokenAmounts,
|
|
feeToken,
|
|
extraArgs,
|
|
blockNumber: event.blockNumber,
|
|
transactionHash: event.transactionHash
|
|
});
|
|
|
|
} catch (error) {
|
|
this.logger.error('Error processing MessageSent event:', error);
|
|
}
|
|
});
|
|
}
|
|
|
|
// Also poll from start block to catch any missed events
|
|
this.pollHistoricalEvents(startBlock);
|
|
}
|
|
|
|
async pollHistoricalEvents(startBlock) {
|
|
while (this.isRunning) {
|
|
try {
|
|
const currentBlock = await this.sourceProvider.getBlockNumber();
|
|
const finalityDelay = this.config.monitoring.finalityDelayBlocks || 0;
|
|
const replayWindow = this.config.monitoring.replayWindowBlocks || 0;
|
|
const toBlock = currentBlock > finalityDelay ? currentBlock - finalityDelay : currentBlock;
|
|
|
|
if (toBlock >= startBlock) {
|
|
this.logger.info(`Polling events from block ${startBlock} to ${toBlock}`);
|
|
|
|
const logs = await this.fetchSourceLogsChunked(startBlock, toBlock);
|
|
this.lastSourcePoll = {
|
|
at: new Date().toISOString(),
|
|
ok: true,
|
|
from_block: startBlock,
|
|
to_block: toBlock,
|
|
logs_fetched: logs.length
|
|
};
|
|
if (this.lastError && this.lastError.scope === 'source_poll') {
|
|
this.lastError = null;
|
|
}
|
|
|
|
this.logger.info(`Fetched ${logs.length} MessageSent log(s) from source router`);
|
|
|
|
for (const log of logs) {
|
|
let decoded;
|
|
try {
|
|
decoded = this.messageSentInterface.parseLog(log);
|
|
} catch (error) {
|
|
this.logger.error('Error decoding MessageSent log:', error);
|
|
continue;
|
|
}
|
|
|
|
const { messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs } = decoded.args;
|
|
|
|
const destSelector = destinationChainSelector.toString();
|
|
const expectedSelector = this.config.destinationChain.chainSelector.toString();
|
|
|
|
if (this.shouldSkipMessageId(messageId)) {
|
|
this.logger.warn(`Skipping historical MessageSent ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
|
|
continue;
|
|
}
|
|
|
|
if (destSelector !== expectedSelector) {
|
|
continue;
|
|
}
|
|
|
|
const scope = this.evaluateMessageScope({ messageId, sender, receiver, tokenAmounts });
|
|
if (!scope.inScope) {
|
|
this.logger.debug(`Ignoring historical message ${messageId}; ${scope.reason}`);
|
|
continue;
|
|
}
|
|
|
|
this.logger.info('Historical MessageSent detected:', {
|
|
messageId,
|
|
destinationChainSelector: destSelector,
|
|
sender,
|
|
targetBridge: scope.targetBridge,
|
|
blockNumber: log.blockNumber,
|
|
transactionHash: log.transactionHash
|
|
});
|
|
this.lastSeenMessage = {
|
|
at: new Date().toISOString(),
|
|
message_id: messageId,
|
|
destination_chain_selector: destSelector,
|
|
sender,
|
|
block_number: log.blockNumber,
|
|
transaction_hash: log.transactionHash
|
|
};
|
|
|
|
if (await this.isDeliveredOnDestination(messageId, scope.targetBridge)) {
|
|
await this.messageQueue.markProcessed(messageId);
|
|
continue;
|
|
}
|
|
|
|
await this.messageQueue.add({
|
|
messageId,
|
|
destinationChainSelector,
|
|
sender,
|
|
receiver,
|
|
targetBridge: scope.targetBridge,
|
|
data,
|
|
tokenAmounts: tokenAmounts.map((ta) => ({
|
|
token: ta.token,
|
|
amount: ta.amount,
|
|
amountType: ta.amountType
|
|
})),
|
|
feeToken,
|
|
extraArgs,
|
|
blockNumber: log.blockNumber,
|
|
transactionHash: log.transactionHash
|
|
}, { forceReplay: true });
|
|
}
|
|
|
|
startBlock = Math.max(0, toBlock - replayWindow + 1);
|
|
}
|
|
|
|
// Wait before next poll (longer interval while relay shedding is on)
|
|
await new Promise(resolve => setTimeout(resolve, this.getSourcePollIntervalMs()));
|
|
|
|
} catch (error) {
|
|
this.logger.error('Error polling historical events:', error);
|
|
this.lastSourcePoll = {
|
|
at: new Date().toISOString(),
|
|
ok: false,
|
|
from_block: startBlock,
|
|
error: RelayService.summarizeError(error)
|
|
};
|
|
this.recordError('source_poll', error, {
|
|
from_block: startBlock
|
|
});
|
|
await new Promise(resolve => setTimeout(resolve, this.getSourcePollIntervalMs()));
|
|
}
|
|
}
|
|
}
|
|
|
|
async startProcessingQueue() {
|
|
const maxConcurrent = this.getMaxConcurrent();
|
|
this.logger.info('Starting message queue processor (maxConcurrent=%s)...', maxConcurrent);
|
|
|
|
const worker = async (workerId) => {
|
|
while (this.isRunning) {
|
|
try {
|
|
if (isRelayShedding()) {
|
|
if (workerId === 0) {
|
|
const stats = this.messageQueue.getStats();
|
|
const now = Date.now();
|
|
if (stats.queueSize > 0 && now - this._lastSheddingLogTs >= 60000) {
|
|
this._lastSheddingLogTs = now;
|
|
this.logger.warn(
|
|
`Relay shedding ON: ${stats.queueSize} message(s) queued; destination delivery paused. ` +
|
|
`Set RELAY_SHEDDING=0 and RELAY_DELIVERY_ENABLED=1 then restart (or reload env) to deliver.`
|
|
);
|
|
}
|
|
}
|
|
await new Promise((r) => setTimeout(r, getRelaySheddingQueueIdleMs()));
|
|
continue;
|
|
}
|
|
|
|
const message = await this.messageQueue.getNext();
|
|
if (!message) {
|
|
await new Promise((resolve) => setTimeout(resolve, 1000));
|
|
continue;
|
|
}
|
|
|
|
this._activeRelayCount += 1;
|
|
try {
|
|
await this.relayMessage(message);
|
|
} catch (error) {
|
|
this.logger.error(`Worker ${workerId} relay error:`, error);
|
|
} finally {
|
|
this._activeRelayCount = Math.max(0, this._activeRelayCount - 1);
|
|
}
|
|
} catch (error) {
|
|
this.logger.error(`Worker ${workerId} queue loop error:`, error);
|
|
await new Promise((resolve) => setTimeout(resolve, 1000));
|
|
}
|
|
}
|
|
};
|
|
|
|
for (let i = 0; i < maxConcurrent; i += 1) {
|
|
void worker(i);
|
|
}
|
|
}
|
|
|
|
async relayMessage(messageData) {
|
|
const { messageId, destinationChainSelector, sender, receiver, data, tokenAmounts } = messageData;
|
|
|
|
if (this.shouldSkipMessageId(messageId)) {
|
|
this.logger.warn(`Skipping queued message ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return null;
|
|
}
|
|
|
|
const scope = this.evaluateMessageScope(messageData);
|
|
if (!scope.inScope) {
|
|
this.logger.info(`Skipping queued message ${messageId}; ${scope.reason}`);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return null;
|
|
}
|
|
|
|
this.logger.info(`Relaying message ${messageId} to destination chain...`);
|
|
|
|
try {
|
|
// On-chain pause (CCIPRelayRouter Pausable): avoid broadcasting reverting txs
|
|
if (this.config.destinationChain.deliveryMode !== 'direct' && this.destinationRelayRouter) {
|
|
try {
|
|
const routerPaused = await this.destinationRelayRouter.paused();
|
|
if (routerPaused) {
|
|
this.logger.warn(
|
|
`Destination relay router is paused; deferring ${messageId} (enable RELAY_SHEDDING=1 to pause off-chain too)`
|
|
);
|
|
await this.messageQueue.resetRetryCount(messageId);
|
|
await this.messageQueue.retry(messageId, { increment: false });
|
|
await new Promise((r) => setTimeout(r, 15000));
|
|
return null;
|
|
}
|
|
} catch (pauseCheckErr) {
|
|
this.logger.debug('Router paused() check skipped or unsupported', pauseCheckErr);
|
|
}
|
|
}
|
|
|
|
// Route to bridge encoded in MessageSent.receiver (bytes). Fallback to static env bridge.
|
|
const targetBridge = scope.targetBridge || this.resolveBridgeForMessage(messageData);
|
|
if (!targetBridge) {
|
|
throw new Error(`No destination bridge for message ${messageId}: receiver decode failed and DEST_RELAY_BRIDGE not set`);
|
|
}
|
|
this.lastRelayAttempt = {
|
|
at: new Date().toISOString(),
|
|
message_id: messageId,
|
|
destination_chain_selector: destinationChainSelector ? destinationChainSelector.toString() : '',
|
|
target_bridge: targetBridge,
|
|
token_count: tokenAmounts.length
|
|
};
|
|
|
|
// Optional allowlist hardening.
|
|
const allowlist = this.getDestinationBridgeAllowlist();
|
|
if (allowlist.length > 0 && !allowlist.includes(String(targetBridge).toLowerCase())) {
|
|
this.logger.info(
|
|
`Skipping message ${messageId} for bridge ${targetBridge}; not in this worker's DEST_RELAY_BRIDGE_ALLOWLIST`
|
|
);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return null;
|
|
}
|
|
|
|
const targetBridgeContract = await this.getDestinationBridgeContract(targetBridge);
|
|
|
|
const alreadyProcessed = await this.isDeliveredOnDestination(messageId, targetBridge);
|
|
if (alreadyProcessed) {
|
|
this.logger.info(`Message ${messageId} already processed on destination; skipping relay tx`);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return null;
|
|
}
|
|
|
|
// Dedicated cW bridge workers can replay historical messages for canonical
|
|
// assets that are no longer configured on the destination multi-token bridge.
|
|
// When that mapping is absent, the destination bridge will always revert with
|
|
// "token not configured", so retire the message instead of retry-looping it.
|
|
if (targetBridgeContract.canonicalToMirrored && data && tokenAmounts.length === 0) {
|
|
try {
|
|
const decodedPayload = ethers.AbiCoder.defaultAbiCoder().decode(
|
|
['address', 'address', 'uint256'],
|
|
data
|
|
);
|
|
const canonicalToken = ethers.getAddress(decodedPayload[0]);
|
|
const mirroredToken = await targetBridgeContract.canonicalToMirrored(canonicalToken);
|
|
if (mirroredToken === ethers.ZeroAddress) {
|
|
this.logger.warn(
|
|
`Skipping queued message ${messageId}; canonical token ${canonicalToken} is not configured on destination bridge ${targetBridge}`
|
|
);
|
|
await this.messageQueue.markProcessed(messageId);
|
|
return null;
|
|
}
|
|
} catch (mappingProbeErr) {
|
|
this.logger.debug(
|
|
`canonicalToMirrored probe skipped for ${messageId}; continuing with relay attempt`,
|
|
mappingProbeErr
|
|
);
|
|
}
|
|
}
|
|
|
|
// Map token addresses from source chain to destination chain
|
|
const mappedTokenAmounts = tokenAmounts.map(ta => {
|
|
const sourceToken = ethers.getAddress(ta.token);
|
|
const destinationToken = this.mapSourceTokenToDestination(sourceToken);
|
|
|
|
this.logger.debug(`Mapping token ${sourceToken} -> ${destinationToken}`);
|
|
|
|
return {
|
|
token: destinationToken, // Map to destination chain token address
|
|
amount: typeof ta.amount === 'bigint' ? ta.amount : BigInt(ta.amount.toString()),
|
|
amountType: Number(ta.amountType) // Ensure it's a number (uint8)
|
|
};
|
|
});
|
|
|
|
let inventoryCheck;
|
|
try {
|
|
inventoryCheck = await this.ensureTargetBridgeInventory(
|
|
messageId,
|
|
targetBridge,
|
|
mappedTokenAmounts
|
|
);
|
|
} catch (inventoryProbeError) {
|
|
this.logger.warn(
|
|
`Bridge inventory probe failed for ${messageId}; deferring message until the next cycle`,
|
|
inventoryProbeError
|
|
);
|
|
this.recordError('bridge_inventory_probe', inventoryProbeError, {
|
|
message_id: messageId,
|
|
target_bridge: targetBridge
|
|
});
|
|
await this.messageQueue.resetRetryCount(messageId);
|
|
await this.messageQueue.retry(messageId, { increment: false });
|
|
await new Promise(resolve => setTimeout(resolve, this.config.retry.retryDelay));
|
|
return null;
|
|
}
|
|
if (!inventoryCheck.ok) {
|
|
const inventoryError = new Error(inventoryCheck.message);
|
|
this.logger.warn(inventoryCheck.message);
|
|
this.recordError('bridge_inventory', inventoryError, {
|
|
message_id: messageId,
|
|
target_bridge: targetBridge,
|
|
token: inventoryCheck.token,
|
|
available_amount: inventoryCheck.availableAmount.toString(),
|
|
required_amount: inventoryCheck.requiredAmount.toString(),
|
|
shortfall: inventoryCheck.shortfall.toString()
|
|
});
|
|
await this.messageQueue.resetRetryCount(messageId);
|
|
await this.messageQueue.retry(messageId, { increment: false });
|
|
await new Promise(resolve => setTimeout(resolve, this.config.retry.retryDelay));
|
|
return null;
|
|
}
|
|
|
|
const normalizedData = this.normalizeRelayData(
|
|
data,
|
|
messageData.receiver,
|
|
tokenAmounts,
|
|
targetBridge,
|
|
sender
|
|
);
|
|
|
|
// Construct Any2EVMMessage struct
|
|
// tokenAmounts is an array of TokenAmount structs: { token: address, amount: uint256, amountType: uint8 }
|
|
const any2EVMMessage = {
|
|
messageId: messageId,
|
|
sourceChainSelector: BigInt(this.config.sourceChainSelector.toString()),
|
|
sender: ethers.AbiCoder.defaultAbiCoder().encode(['address'], [sender]),
|
|
data: normalizedData,
|
|
tokenAmounts: mappedTokenAmounts
|
|
};
|
|
|
|
this.logger.debug('Relaying message with struct:', {
|
|
messageId: any2EVMMessage.messageId,
|
|
sourceChainSelector: any2EVMMessage.sourceChainSelector,
|
|
tokenAmountsCount: any2EVMMessage.tokenAmounts.length
|
|
});
|
|
|
|
const txOptions = await this.getDestinationTxOptions();
|
|
let tx;
|
|
if (this.config.destinationChain.deliveryMode === 'direct') {
|
|
this.logger.info(`Direct-delivery mode: calling bridge ${targetBridge} without relay router`);
|
|
const directBridge = new ethers.Contract(
|
|
targetBridge,
|
|
RelayBridgeABI,
|
|
this.destinationSigner
|
|
);
|
|
tx = await directBridge.ccipReceive(any2EVMMessage, txOptions);
|
|
} else {
|
|
// Call relay router with properly formatted struct
|
|
tx = await this.destinationRelayRouter.relayMessage(
|
|
targetBridge,
|
|
any2EVMMessage,
|
|
txOptions
|
|
);
|
|
}
|
|
|
|
this.logger.info(`Relay transaction sent: ${tx.hash}`);
|
|
|
|
const receipt = await this.waitForDestinationReceipt(tx, messageId);
|
|
|
|
this.logger.info(`Message ${messageId} relayed successfully. Transaction: ${receipt.hash}`);
|
|
this.lastRelaySuccess = {
|
|
at: new Date().toISOString(),
|
|
message_id: messageId,
|
|
destination_chain_selector: destinationChainSelector ? destinationChainSelector.toString() : '',
|
|
target_bridge: targetBridge,
|
|
tx_hash: receipt.hash
|
|
};
|
|
if (
|
|
this.lastError &&
|
|
(
|
|
this.lastError.scope === 'relay_message' ||
|
|
this.lastError.scope === 'bridge_inventory' ||
|
|
this.lastError.scope === 'bridge_inventory_probe'
|
|
)
|
|
) {
|
|
this.lastError = null;
|
|
}
|
|
|
|
// Mark message as processed
|
|
await this.messageQueue.markProcessed(messageId);
|
|
|
|
return receipt;
|
|
|
|
} catch (error) {
|
|
this.logger.error(`Error relaying message ${messageId}:`, error);
|
|
this.recordError('relay_message', error, {
|
|
message_id: messageId
|
|
});
|
|
|
|
// Retry logic
|
|
const retryCount = await this.messageQueue.getRetryCount(messageId);
|
|
if (retryCount < this.config.retry.maxRetries) {
|
|
this.logger.info(`Retrying message ${messageId} (attempt ${retryCount + 1})`);
|
|
await this.messageQueue.retry(messageId);
|
|
await new Promise(resolve => setTimeout(resolve, this.config.retry.retryDelay));
|
|
} else {
|
|
this.logger.error(`Message ${messageId} failed after ${this.config.retry.maxRetries} retries`);
|
|
await this.messageQueue.markFailed(messageId);
|
|
}
|
|
|
|
throw error;
|
|
}
|
|
}
|
|
}
|