poanetwork
/
tokenbridge-monitor
mirror of https://github.com/poanetwork/tokenbridge-monitor.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Support erc-to-native bridge

This commit is contained in:
Kirill Fedoseev 2022-04-02 15:49:18 +04:00
parent 840328e120
commit a652803973
26 changed files with 2243 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
config.schema.json
View File

@ -58,6 +58,9 @@
"additionalProperties": {
"type": "object",
"properties": {
"is_erc_to_native": {
"type": "boolean"
},
"home": {
"$ref": "#/$defs/side_config"
},
@ -93,6 +96,15 @@
},
"different_information_signatures": {
"$ref": "#/$defs/alert_config"
},
"unknown_erc_to_native_message_confirmation": {
"$ref": "#/$defs/alert_config"
},
"unknown_erc_to_native_message_execution": {
"$ref": "#/$defs/alert_config"
},
"stuck_erc_to_native_message_confirmation": {
"$ref": "#/$defs/alert_config"
}
},
"additionalProperties": false
@ -238,6 +250,36 @@
"type": "string",
"pattern": "^0x[a-fA-F0-9]{40}$"
}
},
"erc_to_native_tokens": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"properties": {
"address": {
"type": "string",
"pattern": "^0x[a-fA-F0-9]{40}$"
},
"start_block": {
"type": "integer"
},
"end_block": {
"type": "integer"
}
},
"required": [
"address"
]
}
},
"erc_to_native_blacklisted_senders": {
"type": "array",
"minItems": 1,
"items": {
"type": "string",
"pattern": "^0x[a-fA-F0-9]{40}$"
}
}
},
"required": [

36
config.yml
View File

@ -26,15 +26,16 @@ chains:
block_index_interval: 60s
xdai:
rpc:
host: https://dai.poa.network/oe-only
host: https://rpc.xdaichain.com/oe-only
timeout: 20s
rps: 10
chain_id: 100
block_time: 5s
block_index_interval: 30s
safe_logs_request: true
poa:
rpc:
host: https://core.poa.network
host: https://core.poanetwork.dev
timeout: 20s
rps: 10
chain_id: 99
@ -57,6 +58,35 @@ chains:
block_time: 15s
block_index_interval: 60s
bridges:
xdai:
is_erc_to_native: true
home:
chain: xdai
address: 0x7301CFA0e1756B71869E93d4e4Dca5c7d0eb0AA6
validator_contract_address: 0xB289f0e6fBDFf8EEE340498a56e1787B303F1B6D
start_block: 756
required_block_confirmations: 12
max_block_range_size: 2000
foreign:
chain: mainnet
address: 0x4aa42145Aa6Ebf72e164C9bBC74fbD3788045016
validator_contract_address: 0xe1579dEbdD2DF16Ebdb9db8694391fa74EeA201E
start_block: 6478411
required_block_confirmations: 12
max_block_range_size: 1000
erc_to_native_tokens:
- address: 0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359
start_block: 6478411
end_block: 9884448
- address: 0x6B175474E89094C44Da98b954EedeAC495271d0F
start_block: 8928158
erc_to_native_blacklisted_senders:
- 0x0000000000000000000000000000000000000000
- 0x5d3a536E4D6DbD6114cc1Ead35777bAB948E3643
alerts:
unknown_erc_to_native_message_confirmation:
unknown_erc_to_native_message_execution:
stuck_erc_to_native_message_confirmation:
xdai-amb:
home:
chain: xdai
@ -92,7 +122,7 @@ bridges:
chain: sokol
address: 0xFe446bEF1DbF7AFE24E81e05BC8B271C1BA9a560
validator_contract_address: 0x0c7A04cc9B1fF1184c5bc7253869727f29593465
start_block: 9849619
start_block: 9849617
required_block_confirmations: 12
max_block_range_size: 10000
foreign:

35
config/config.go
View File

@ -31,16 +31,24 @@ type ReloadJobConfig struct {
EndBlock uint `yaml:"end_block"`
}
type TokenConfig struct {
Address common.Address `yaml:"address"`
StartBlock uint `yaml:"start_block"`
EndBlock uint `yaml:"end_block"`
}
type BridgeSideConfig struct {
ChainName string `yaml:"chain"`
Chain *ChainConfig `yaml:"-"`
Address common.Address `yaml:"address"`
ValidatorContractAddress common.Address `yaml:"validator_contract_address"`
StartBlock uint `yaml:"start_block"`
BlockConfirmations uint `yaml:"required_block_confirmations"`
MaxBlockRangeSize uint `yaml:"max_block_range_size"`
RefetchEvents []*ReloadJobConfig `yaml:"refetch_events"`
WhitelistedSenders []common.Address `yaml:"whitelisted_senders"`
ChainName string `yaml:"chain"`
Chain *ChainConfig `yaml:"-"`
Address common.Address `yaml:"address"`
ValidatorContractAddress common.Address `yaml:"validator_contract_address"`
StartBlock uint `yaml:"start_block"`
BlockConfirmations uint `yaml:"required_block_confirmations"`
MaxBlockRangeSize uint `yaml:"max_block_range_size"`
RefetchEvents []*ReloadJobConfig `yaml:"refetch_events"`
WhitelistedSenders []common.Address `yaml:"whitelisted_senders"`
ErcToNativeTokens []TokenConfig `yaml:"erc_to_native_tokens"`
ErcToNativeBlacklistedSenders []common.Address `yaml:"erc_to_native_blacklisted_senders"`
}
type BridgeAlertConfig struct {
@ -49,10 +57,11 @@ type BridgeAlertConfig struct {
}
type BridgeConfig struct {
ID string `yaml:"-"`
Home *BridgeSideConfig `yaml:"home"`
Foreign *BridgeSideConfig `yaml:"foreign"`
Alerts map[string]*BridgeAlertConfig `yaml:"alerts"`
ID string `yaml:"-"`
IsErcToNative bool `yaml:"is_erc_to_native"`
Home *BridgeSideConfig `yaml:"home"`
Foreign *BridgeSideConfig `yaml:"foreign"`
Alerts map[string]*BridgeAlertConfig `yaml:"alerts"`
}
type DBConfig struct {

12
contract/constants/abi.go
View File

@ -2,14 +2,18 @@ package constants
import (
_ "embed"
"github.com/ethereum/go-ethereum/accounts/abi"
"strings"
"github.com/ethereum/go-ethereum/accounts/abi"
)
//go:embed amb.json
var ambJsonABI string
var AMB abi.ABI
//go:embed erc_to_native.json
var etnJsonABI string
var AMB, ERC_TO_NATIVE abi.ABI
func init() {
var err error
@ -17,4 +21,8 @@ func init() {
if err != nil {
panic(err)
}
ERC_TO_NATIVE, err = abi.JSON(strings.NewReader(etnJsonABI))
if err != nil {
panic(err)
}
}

196
contract/constants/erc_to_native.json Normal file
View File

@ -0,0 +1,196 @@
[
{
"constant": true,
"inputs": [],
"name": "validatorContract",
"outputs": [
{
"name": "",
"type": "address"
}
],
"payable": false,
"stateMutability": "view",
"type": "function"
},
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"name": "recipient",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
},
{
"indexed": false,
"name": "transactionHash",
"type": "bytes32"
}
],
"name": "RelayedMessage",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "from",
"type": "address"
},
{
"indexed": true,
"name": "to",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
}
],
"name": "Transfer",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"name": "recipient",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
}
],
"name": "UserRequestForSignature",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"name": "recipient",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
},
{
"indexed": false,
"name": "transactionHash",
"type": "bytes32"
}
],
"name": "AffirmationCompleted",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "signer",
"type": "address"
},
{
"indexed": false,
"name": "messageHash",
"type": "bytes32"
}
],
"name": "SignedForUserRequest",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "signer",
"type": "address"
},
{
"indexed": false,
"name": "transactionHash",
"type": "bytes32"
}
],
"name": "SignedForAffirmation",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"name": "authorityResponsibleForRelay",
"type": "address"
},
{
"indexed": false,
"name": "messageHash",
"type": "bytes32"
},
{
"indexed": false,
"name": "NumberOfCollectedSignatures",
"type": "uint256"
}
],
"name": "CollectedSignatures",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"name": "recipient",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
}
],
"name": "UserRequestForAffirmation",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "validator",
"type": "address"
}
],
"name": "ValidatorAdded",
"type": "event"
},
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "validator",
"type": "address"
}
],
"name": "ValidatorRemoved",
"type": "event"
}
]

3
db/migrations/000004_erc_to_native_messages.down.sql Normal file
View File

@ -0,0 +1,3 @@
DROP INDEX erc_to_native_messages_bridge_id_msg_hash_idx;
DROP TABLE erc_to_native_messages;
DROP DOMAIN UINT;

15
db/migrations/000004_erc_to_native_messages.up.sql Normal file
View File

@ -0,0 +1,15 @@
CREATE DOMAIN UINT AS DECIMAL(80, 0) NOT NULL DEFAULT 0;
CREATE TABLE erc_to_native_messages
(
id SERIAL PRIMARY KEY,
bridge_id TEXT_ID,
msg_hash WORD,
direction DIRECTION,
receiver ADDRESS,
value UINT,
updated_at TS_NOW,
created_at TS_NOW
);
CREATE UNIQUE INDEX erc_to_native_messages_bridge_id_msg_hash_idx ON erc_to_native_messages (bridge_id, msg_hash);
DROP INDEX sent_messages_bridge_id_msg_hash_idx;
CREATE INDEX sent_messages_bridge_id_msg_hash_idx ON sent_messages (bridge_id, msg_hash);

28
db/queries.sql
View File

@ -1,28 +0,0 @@
-- message stats
-- total
SELECT count(*)
FROM messages
WHERE bridge_id = '?'
AND direction = 'home_to_foreign';
SELECT count(*)
FROM messages
WHERE bridge_id = '?'
AND direction = 'foreign_to_home';
-- foreign to home pending
SELECT m.*
FROM messages m
LEFT JOIN executed_messages em ON m.id = em.message_id AND m.bridge_id = em.bridge_id
WHERE m.bridge_id = '?'
AND m.direction = 'foreign_to_home'
AND em.log_id IS NULL;
-- home to foreign oracle-driven pending
SELECT m.*
FROM messages m
LEFT JOIN executed_messages em ON m.id = em.message_id AND m.bridge_id = em.bridge_id
WHERE m.bridge_id = '?'
AND m.direction = 'home_to_foreign'
AND m.data_type = 0
AND em.log_id IS NULL;

24
entity/erc_to_native_message.go Normal file
View File

@ -0,0 +1,24 @@
package entity
import (
"context"
"time"
"github.com/ethereum/go-ethereum/common"
)
type ErcToNativeMessage struct {
ID uint `db:"id"`
BridgeID string `db:"bridge_id"`
MsgHash common.Hash `db:"msg_hash"`
Direction Direction `db:"direction"`
Receiver common.Address `db:"receiver"`
Value string `db:"value"`
CreatedAt *time.Time `db:"created_at"`
UpdatedAt *time.Time `db:"updated_at"`
}
type ErcToNativeMessagesRepo interface {
Ensure(ctx context.Context, msg *ErcToNativeMessage) error
FindByMsgHash(ctx context.Context, bridgeID string, msgHash common.Hash) (*ErcToNativeMessage, error)
}

2
entity/log.go
View File

@ -26,7 +26,7 @@ type Log struct {
type LogsRepo interface {
Ensure(ctx context.Context, logs ...*Log) error
GetByID(ctx context.Context, id uint) (*Log, error)
FindByBlockRange(ctx context.Context, chainID string, addr common.Address, fromBlock, toBlock uint) ([]*Log, error)
FindByBlockRange(ctx context.Context, chainID string, addr []common.Address, fromBlock, toBlock uint) ([]*Log, error)
FindByBlockNumber(ctx context.Context, chainID string, block uint) ([]*Log, error)
FindByTxHash(ctx context.Context, txHash common.Hash) ([]*Log, error)
}

10
ethclient/ethclient.go
View File

@ -135,6 +135,16 @@ func (c *Client) TransactionByHash(ctx context.Context, txHash common.Hash) (*ty
return tx, err
}
func (c *Client) TransactionReceiptByHash(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
defer ObserveDuration(c.ChainID, c.url, "eth_getTransactionReceipt")()
ctx, cancel := context.WithTimeout(ctx, c.timeout)
defer cancel()
receipt, err := c.client.TransactionReceipt(ctx, txHash)
ObserveError(c.ChainID, c.url, "eth_getTransactionReceipt", err)
return receipt, err
}
func (c *Client) CallContract(ctx context.Context, msg ethereum.CallMsg) ([]byte, error) {
defer ObserveDuration(c.ChainID, c.url, "eth_call")()
ctx, cancel := context.WithTimeout(ctx, c.timeout)

1185
grafana/dashboards/xdai.json Normal file
View File

File diff suppressed because it is too large Load Diff

21
monitor/alerts/alert_manager.go
View File

@ -83,6 +83,27 @@ func NewAlertManager(logger logging.Logger, db *db.DB, cfg *config.BridgeConfig)
Func: provider.FindDifferentInformationSignatures,
Metric: NewAlertDifferentInformationSignatures(cfg.ID),
}
case "unknown_erc_to_native_message_confirmation":
jobs[name] = &Job{
Interval: time.Minute,
Timeout: time.Second * 10,
Func: provider.FindUnknownErcToNativeConfirmations,
Metric: NewAlertUnknownErcToNativeMessageConfirmation(cfg.ID),
}
case "unknown_erc_to_native_message_execution":
jobs[name] = &Job{
Interval: time.Minute,
Timeout: time.Second * 10,
Func: provider.FindUnknownErcToNativeExecutions,
Metric: NewAlertUnknownErcToNativeMessageExecution(cfg.ID),
}
case "stuck_erc_to_native_message_confirmation":
jobs[name] = &Job{
Interval: time.Minute * 5,
Timeout: time.Second * 20,
Func: provider.FindStuckErcToNativeMessages,
Metric: NewAlertStuckErcToNativeMessageConfirmation(cfg.ID),
}
default:
return nil, fmt.Errorf("unknown alert type %q", name)
}

169
monitor/alerts/db_alerts_provider.go
View File

@ -569,3 +569,172 @@ func (p *DBAlertsProvider) FindUnknownInformationExecutions(ctx context.Context,
}
return alerts, nil
}
func (p *DBAlertsProvider) FindUnknownErcToNativeConfirmations(ctx context.Context, params *AlertJobParams) ([]AlertValues, error) {
minProcessedTS, err := p.findMinProcessedTime(ctx, params)
if err != nil {
return nil, err
}
q, args, err := sq.Select("l.chain_id", "l.block_number", "l.transaction_hash", "sm.signer", "sm.msg_hash", "EXTRACT(EPOCH FROM now() - bt.timestamp)::int as age").
From("signed_messages sm").
Join("logs l ON l.id = sm.log_id").
Join("block_timestamps bt on bt.chain_id = l.chain_id AND bt.block_number = l.block_number").
LeftJoin("erc_to_native_messages m ON sm.bridge_id = m.bridge_id AND m.msg_hash = sm.msg_hash").
Where(sq.Eq{"m.id": nil, "sm.bridge_id": params.Bridge, "l.chain_id": params.HomeChainID}).
Where(sq.GtOrEq{"l.block_number": params.HomeStartBlockNumber}).
Where(sq.LtOrEq{"bt.timestamp": minProcessedTS}).
PlaceholderFormat(sq.Dollar).
ToSql()
if err != nil {
return nil, fmt.Errorf("can't build query: %w", err)
}
res := make([]UnknownConfirmation, 0, 5)
err = p.db.SelectContext(ctx, &res, q, args...)
if err != nil {
return nil, fmt.Errorf("can't select alerts: %w", err)
}
alerts := make([]AlertValues, len(res))
for i := range res {
alerts[i] = res[i].AlertValues()
}
return alerts, nil
}
type UnknownErcToNativeExecution struct {
ChainID string `db:"chain_id"`
BlockNumber uint64 `db:"block_number"`
Age time.Duration `db:"age"`
TransactionHash common.Hash `db:"transaction_hash"`
MsgHash common.Hash `db:"msg_hash"`
}
func (c *UnknownErcToNativeExecution) AlertValues() AlertValues {
return AlertValues{
Labels: map[string]string{
"chain_id": c.ChainID,
"block_number": strconv.FormatUint(c.BlockNumber, 10),
"tx_hash": c.TransactionHash.String(),
"msg_hash": c.MsgHash.String(),
},
Value: float64(c.Age),
}
}
func (p *DBAlertsProvider) FindUnknownErcToNativeExecutions(ctx context.Context, params *AlertJobParams) ([]AlertValues, error) {
minProcessedTS, err := p.findMinProcessedTime(ctx, params)
if err != nil {
return nil, err
}
q, args, err := sq.Select("l.chain_id", "l.block_number", "l.transaction_hash", "em.message_id as msg_hash", "EXTRACT(EPOCH FROM now() - bt.timestamp)::int as age").
From("executed_messages em").
Join("logs l ON l.id = em.log_id").
Join("block_timestamps bt on bt.chain_id = l.chain_id AND bt.block_number = l.block_number").
LeftJoin("erc_to_native_messages m ON em.bridge_id = m.bridge_id AND em.message_id = m.msg_hash").
Where(sq.Eq{"m.id": nil, "em.bridge_id": params.Bridge}).
Where(sq.Or{
sq.And{
sq.Eq{"l.chain_id": params.HomeChainID},
sq.GtOrEq{"l.block_number": params.HomeStartBlockNumber},
},
sq.And{
sq.Eq{"l.chain_id": params.ForeignChainID},
sq.GtOrEq{"l.block_number": params.ForeignStartBlockNumber},
},
}).
Where(sq.LtOrEq{"bt.timestamp": minProcessedTS}).
PlaceholderFormat(sq.Dollar).
ToSql()
if err != nil {
return nil, fmt.Errorf("can't build query: %w", err)
}
res := make([]UnknownErcToNativeExecution, 0, 5)
err = p.db.SelectContext(ctx, &res, q, args...)
if err != nil {
return nil, fmt.Errorf("can't select alerts: %w", err)
}
alerts := make([]AlertValues, len(res))
for i := range res {
alerts[i] = res[i].AlertValues()
}
return alerts, nil
}
type StuckErcToNativeMessage struct {
ChainID string `db:"chain_id"`
BlockNumber uint64 `db:"block_number"`
Age time.Duration `db:"age"`
TransactionHash common.Hash `db:"transaction_hash"`
MsgHash common.Hash `db:"msg_hash"`
Count uint64 `db:"count"`
Receiver common.Address `db:"receiver"`
Value string `db:"value"`
}
func (c *StuckErcToNativeMessage) AlertValues() AlertValues {
return AlertValues{
Labels: map[string]string{
"chain_id": c.ChainID,
"block_number": strconv.FormatUint(c.BlockNumber, 10),
"tx_hash": c.TransactionHash.String(),
"msg_hash": c.MsgHash.String(),
"count": strconv.FormatUint(c.Count, 10),
"receiver": c.Receiver.String(),
"value": c.Value,
},
Value: float64(c.Age),
}
}
func (p *DBAlertsProvider) FindStuckErcToNativeMessages(ctx context.Context, params *AlertJobParams) ([]AlertValues, error) {
query := `
SELECT l.chain_id,
l.block_number,
l.transaction_hash,
sm.msg_hash,
count(s.log_id) as count,
EXTRACT(EPOCH FROM now() - ts.timestamp)::int as age,
m.receiver,
m.value / 1e18 as value
FROM sent_messages sm
JOIN logs l on l.id = sm.log_id
JOIN block_timestamps ts on ts.chain_id = l.chain_id AND ts.block_number = l.block_number
JOIN erc_to_native_messages m on sm.bridge_id = m.bridge_id AND m.msg_hash = sm.msg_hash
LEFT JOIN signed_messages s on s.bridge_id = m.bridge_id AND m.msg_hash = s.msg_hash
LEFT JOIN collected_messages cm on m.bridge_id = cm.bridge_id AND cm.msg_hash = m.msg_hash
WHERE m.direction::direction_enum = 'home_to_foreign'
AND cm.log_id IS NULL
AND sm.bridge_id = $1
AND l.block_number >= $2
GROUP BY sm.log_id, l.id, ts.timestamp, m.id
UNION
SELECT l.chain_id,
l.block_number,
l.transaction_hash,
sm.msg_hash,
count(s.log_id) as count,
EXTRACT(EPOCH FROM now() - ts.timestamp)::int as age,
m.receiver,
m.value / 1e18 as value
FROM sent_messages sm
JOIN logs l on l.id = sm.log_id
JOIN block_timestamps ts on ts.chain_id = l.chain_id AND ts.block_number = l.block_number
JOIN erc_to_native_messages m on sm.bridge_id = m.bridge_id AND m.msg_hash = sm.msg_hash
LEFT JOIN signed_messages s on s.bridge_id = m.bridge_id AND m.msg_hash = s.msg_hash
LEFT JOIN executed_messages em on m.bridge_id = em.bridge_id AND em.message_id = m.msg_hash
WHERE m.direction::direction_enum = 'foreign_to_home'
AND em.log_id IS NULL
AND sm.bridge_id = $1
AND l.block_number >= $3
AND m.value > 0
GROUP BY sm.log_id, l.id, ts.timestamp, m.id`
res := make([]StuckErcToNativeMessage, 0, 5)
err := p.db.SelectContext(ctx, &res, query, params.Bridge, params.HomeStartBlockNumber, params.ForeignStartBlockNumber)
if err != nil {
return nil, fmt.Errorf("can't select alerts: %w", err)
}
alerts := make([]AlertValues, len(res))
for i := range res {
alerts[i] = res[i].AlertValues()
}
return alerts, nil
}

27
monitor/alerts/metrics.go
View File

@ -87,4 +87,31 @@ var (
ConstLabels: prometheus.Labels{"bridge_id": bridge},
}, []string{"chain_id", "block_number", "tx_hash", "message_id", "count"})
}
NewAlertUnknownErcToNativeMessageConfirmation = func(bridge string) *prometheus.GaugeVec {
return promauto.NewGaugeVec(prometheus.GaugeOpts{
Namespace: "alert",
Subsystem: "monitor",
Name: "unknown_erc_to_native_message_confirmation",
Help: "Shows found unknown ERC_TO_NATIVE message confirmation sent by some validator.",
ConstLabels: prometheus.Labels{"bridge_id": bridge},
}, []string{"chain_id", "block_number", "tx_hash", "signer", "msg_hash"})
}
NewAlertUnknownErcToNativeMessageExecution = func(bridge string) *prometheus.GaugeVec {
return promauto.NewGaugeVec(prometheus.GaugeOpts{
Namespace: "alert",
Subsystem: "monitor",
Name: "unknown_erc_to_native_message_execution",
Help: "Shows found unknown ERC_TO_NATIVE message execution.",
ConstLabels: prometheus.Labels{"bridge_id": bridge},
}, []string{"chain_id", "block_number", "tx_hash", "msg_hash"})
}
NewAlertStuckErcToNativeMessageConfirmation = func(bridge string) *prometheus.GaugeVec {
return promauto.NewGaugeVec(prometheus.GaugeOpts{
Namespace: "alert",
Subsystem: "monitor",
Name: "stuck_erc_to_native_message_confirmation",
Help: "Shows ERC_TO_NATIVE message for which signatures are still in the pending state.",
ConstLabels: prometheus.Labels{"bridge_id": bridge},
}, []string{"chain_id", "block_number", "tx_hash", "msg_hash", "count", "receiver", "value"})
}
)

171
monitor/handlers.go
View File

@ -1,6 +1,8 @@
package monitor
import (
"amb-monitor/config"
"amb-monitor/contract/constants"
"amb-monitor/entity"
"amb-monitor/ethclient"
"amb-monitor/repository"
@ -9,21 +11,26 @@ import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
)
type EventHandler func(ctx context.Context, log *entity.Log, data map[string]interface{}) error
type BridgeEventHandler struct {
repo *repository.Repo
bridgeID string
homeClient *ethclient.Client
repo *repository.Repo
bridgeID string
homeClient *ethclient.Client
foreignClient *ethclient.Client
cfg *config.BridgeConfig
}
func NewBridgeEventHandler(repo *repository.Repo, bridgeID string, homeClient *ethclient.Client) *BridgeEventHandler {
func NewBridgeEventHandler(repo *repository.Repo, bridgeID string, homeClient, foreignClient *ethclient.Client, cfg *config.BridgeConfig) *BridgeEventHandler {
return &BridgeEventHandler{
repo: repo,
bridgeID: bridgeID,
homeClient: homeClient,
repo: repo,
bridgeID: bridgeID,
homeClient: homeClient,
foreignClient: foreignClient,
cfg: cfg,
}
}
@ -56,6 +63,77 @@ func (p *BridgeEventHandler) HandleLegacyUserRequestForAffirmation(ctx context.C
})
}
func (p *BridgeEventHandler) HandleErcToNativeTransfer(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
from := data["from"].(common.Address)
value := data["value"].(*big.Int)
for _, addr := range p.cfg.Foreign.ErcToNativeBlacklistedSenders {
if from == addr {
return nil
}
}
receipt, err := p.foreignClient.TransactionReceiptByHash(ctx, log.TransactionHash)
if err != nil {
return fmt.Errorf("failed to get transaction receipt by hash %s: %w", log.TransactionHash, err)
}
for _, l := range receipt.Logs {
if len(l.Topics) > 0 && l.Topics[0] == constants.ERC_TO_NATIVE.Events["UserRequestForAffirmation"].ID {
return nil
}
}
valueBytes := common.BigToHash(value)
msg := from[:]
msg = append(msg, valueBytes[:]...)
msg = append(msg, log.TransactionHash[:]...)
msgHash := crypto.Keccak256Hash(msg)
message := &entity.ErcToNativeMessage{
BridgeID: p.bridgeID,
Direction: entity.DirectionForeignToHome,
MsgHash: msgHash,
Receiver: from,
Value: value.String(),
}
err = p.repo.ErcToNativeMessages.Ensure(ctx, message)
if err != nil {
return err
}
return p.repo.SentMessages.Ensure(ctx, &entity.SentMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MsgHash: msgHash,
})
}
func (p *BridgeEventHandler) HandleErcToNativeUserRequestForAffirmation(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
recipient := data["recipient"].(common.Address)
value := data["value"].(*big.Int)
valueBytes := common.BigToHash(value)
msg := recipient[:]
msg = append(msg, valueBytes[:]...)
msg = append(msg, log.TransactionHash[:]...)
msgHash := crypto.Keccak256Hash(msg)
message := &entity.ErcToNativeMessage{
BridgeID: p.bridgeID,
Direction: entity.DirectionForeignToHome,
MsgHash: msgHash,
Receiver: recipient,
Value: value.String(),
}
err := p.repo.ErcToNativeMessages.Ensure(ctx, message)
if err != nil {
return err
}
return p.repo.SentMessages.Ensure(ctx, &entity.SentMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MsgHash: msgHash,
})
}
func (p *BridgeEventHandler) HandleUserRequestForSignature(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
encodedData := data["encodedData"].([]byte)
message := unmarshalMessage(p.bridgeID, entity.DirectionHomeToForeign, encodedData)
@ -85,6 +163,35 @@ func (p *BridgeEventHandler) HandleLegacyUserRequestForSignature(ctx context.Con
})
}
func (p *BridgeEventHandler) HandleErcToNativeUserRequestForSignature(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
recipient := data["recipient"].(common.Address)
value := data["value"].(*big.Int)
valueBytes := common.BigToHash(value)
msg := recipient[:]
msg = append(msg, valueBytes[:]...)
msg = append(msg, log.TransactionHash[:]...)
msg = append(msg, p.cfg.Foreign.Address[:]...)
msgHash := crypto.Keccak256Hash(msg)
message := &entity.ErcToNativeMessage{
BridgeID: p.bridgeID,
Direction: entity.DirectionHomeToForeign,
MsgHash: msgHash,
Receiver: recipient,
Value: value.String(),
}
err := p.repo.ErcToNativeMessages.Ensure(ctx, message)
if err != nil {
return err
}
return p.repo.SentMessages.Ensure(ctx, &entity.SentMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MsgHash: msgHash,
})
}
func (p *BridgeEventHandler) HandleSignedForUserRequest(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
msgHash := data["messageHash"].([32]byte)
validator := data["signer"].(common.Address)
@ -97,14 +204,19 @@ func (p *BridgeEventHandler) HandleSignedForUserRequest(ctx context.Context, log
})
}
func (p *BridgeEventHandler) HandleSignedForAffirmation(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
msgHash := data["messageHash"].([32]byte)
func (p *BridgeEventHandler) HandleErcToNativeSignedForAffirmation(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
validator := data["signer"].(common.Address)
tx, err := p.homeClient.TransactionByHash(ctx, log.TransactionHash)
if err != nil {
return fmt.Errorf("failed to get transaction by hash %s: %w", log.TransactionHash, err)
}
msg := tx.Data()[16:]
return p.repo.SignedMessages.Ensure(ctx, &entity.SignedMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MsgHash: msgHash,
MsgHash: crypto.Keccak256Hash(msg),
Signer: validator,
})
}
@ -121,6 +233,26 @@ func (p *BridgeEventHandler) HandleRelayedMessage(ctx context.Context, log *enti
})
}
func (p *BridgeEventHandler) HandleErcToNativeRelayedMessage(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
recipient := data["recipient"].(common.Address)
value := data["value"].(*big.Int)
transactionHash := data["transactionHash"].([32]byte)
valueBytes := common.BigToHash(value)
msg := recipient[:]
msg = append(msg, valueBytes[:]...)
msg = append(msg, transactionHash[:]...)
msg = append(msg, log.Address[:]...)
return p.repo.ExecutedMessages.Ensure(ctx, &entity.ExecutedMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MessageID: crypto.Keccak256Hash(msg),
Status: true,
})
}
func (p *BridgeEventHandler) HandleAffirmationCompleted(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
messageID := data["messageId"].([32]byte)
status := data["status"].(bool)
@ -133,6 +265,25 @@ func (p *BridgeEventHandler) HandleAffirmationCompleted(ctx context.Context, log
})
}
func (p *BridgeEventHandler) HandleErcToNativeAffirmationCompleted(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
recipient := data["recipient"].(common.Address)
value := data["value"].(*big.Int)
transactionHash := data["transactionHash"].([32]byte)
valueBytes := common.BigToHash(value)
msg := recipient[:]
msg = append(msg, valueBytes[:]...)
msg = append(msg, transactionHash[:]...)
return p.repo.ExecutedMessages.Ensure(ctx, &entity.ExecutedMessage{
LogID: log.ID,
BridgeID: p.bridgeID,
MessageID: crypto.Keccak256Hash(msg),
Status: true,
})
}
func (p *BridgeEventHandler) HandleCollectedSignatures(ctx context.Context, log *entity.Log, data map[string]interface{}) error {
msgHash := data["messageHash"].([32]byte)
relayer := data["authorityResponsibleForRelay"].(common.Address)

122
monitor/monitor.go
View File

@ -30,6 +30,7 @@ import (
)
type ContractMonitor struct {
bridgeCfg *config.BridgeConfig
cfg *config.BridgeSideConfig
logger logging.Logger
repo *repository.Repo
@ -57,12 +58,16 @@ type Monitor struct {
const defaultSyncedThreshold = 10
func newContractMonitor(ctx context.Context, logger logging.Logger, repo *repository.Repo, bridge string, cfg *config.BridgeSideConfig) (*ContractMonitor, error) {
func newContractMonitor(ctx context.Context, logger logging.Logger, repo *repository.Repo, bridgeCfg *config.BridgeConfig, cfg *config.BridgeSideConfig) (*ContractMonitor, error) {
client, err := ethclient.NewClient(cfg.Chain.RPC.Host, cfg.Chain.RPC.Timeout, cfg.Chain.ChainID)
if err != nil {
return nil, fmt.Errorf("failed to start eth client: %w", err)
}
bridgeContract := contract.NewContract(client, cfg.Address, constants.AMB)
abi := constants.AMB
if bridgeCfg.IsErcToNative {
abi = constants.ERC_TO_NATIVE
}
bridgeContract := contract.NewContract(client, cfg.Address, abi)
if cfg.ValidatorContractAddress == (common.Address{}) {
cfg.ValidatorContractAddress, err = bridgeContract.ValidatorContractAddress(ctx)
if err != nil {
@ -94,12 +99,13 @@ func newContractMonitor(ctx context.Context, logger logging.Logger, repo *reposi
}
}
commonLabels := prometheus.Labels{
"bridge_id": bridge,
"bridge_id": bridgeCfg.ID,
"chain_id": client.ChainID,
"address": cfg.Address.String(),
}
return &ContractMonitor{
logger: logger,
bridgeCfg: bridgeCfg,
cfg: cfg,
repo: repo,
client: client,
@ -117,11 +123,11 @@ func newContractMonitor(ctx context.Context, logger logging.Logger, repo *reposi
func NewMonitor(ctx context.Context, logger logging.Logger, dbConn *db.DB, repo *repository.Repo, cfg *config.BridgeConfig) (*Monitor, error) {
logger.Info("initializing bridge monitor")
homeMonitor, err := newContractMonitor(ctx, logger.WithField("contract", "home"), repo, cfg.ID, cfg.Home)
homeMonitor, err := newContractMonitor(ctx, logger.WithField("contract", "home"), repo, cfg, cfg.Home)
if err != nil {
return nil, fmt.Errorf("failed to initialize home side monitor: %w", err)
}
foreignMonitor, err := newContractMonitor(ctx, logger.WithField("contract", "foreign"), repo, cfg.ID, cfg.Foreign)
foreignMonitor, err := newContractMonitor(ctx, logger.WithField("contract", "foreign"), repo, cfg, cfg.Foreign)
if err != nil {
return nil, fmt.Errorf("failed to initialize foreign side monitor: %w", err)
}
@ -129,23 +135,32 @@ func NewMonitor(ctx context.Context, logger logging.Logger, dbConn *db.DB, repo
if err != nil {
return nil, fmt.Errorf("failed to initialize alert manager: %w", err)
}
handlers := NewBridgeEventHandler(repo, cfg.ID, homeMonitor.client)
homeMonitor.eventHandlers["UserRequestForSignature"] = handlers.HandleUserRequestForSignature
homeMonitor.eventHandlers["UserRequestForSignature0"] = handlers.HandleLegacyUserRequestForSignature
handlers := NewBridgeEventHandler(repo, cfg.ID, homeMonitor.client, foreignMonitor.client, cfg)
if cfg.IsErcToNative {
homeMonitor.eventHandlers["UserRequestForSignature"] = handlers.HandleErcToNativeUserRequestForSignature
homeMonitor.eventHandlers["SignedForAffirmation"] = handlers.HandleErcToNativeSignedForAffirmation
homeMonitor.eventHandlers["AffirmationCompleted"] = handlers.HandleErcToNativeAffirmationCompleted
foreignMonitor.eventHandlers["UserRequestForAffirmation"] = handlers.HandleErcToNativeUserRequestForAffirmation
foreignMonitor.eventHandlers["Transfer"] = handlers.HandleErcToNativeTransfer
foreignMonitor.eventHandlers["RelayedMessage"] = handlers.HandleErcToNativeRelayedMessage
} else {
homeMonitor.eventHandlers["UserRequestForSignature"] = handlers.HandleUserRequestForSignature
homeMonitor.eventHandlers["UserRequestForSignature0"] = handlers.HandleLegacyUserRequestForSignature
homeMonitor.eventHandlers["SignedForAffirmation"] = handlers.HandleSignedForUserRequest
homeMonitor.eventHandlers["AffirmationCompleted"] = handlers.HandleAffirmationCompleted
homeMonitor.eventHandlers["AffirmationCompleted0"] = handlers.HandleAffirmationCompleted
homeMonitor.eventHandlers["UserRequestForInformation"] = handlers.HandleUserRequestForInformation
homeMonitor.eventHandlers["SignedForInformation"] = handlers.HandleSignedForInformation
homeMonitor.eventHandlers["InformationRetrieved"] = handlers.HandleInformationRetrieved
foreignMonitor.eventHandlers["UserRequestForAffirmation"] = handlers.HandleUserRequestForAffirmation
foreignMonitor.eventHandlers["UserRequestForAffirmation0"] = handlers.HandleLegacyUserRequestForAffirmation
foreignMonitor.eventHandlers["RelayedMessage"] = handlers.HandleRelayedMessage
foreignMonitor.eventHandlers["RelayedMessage0"] = handlers.HandleRelayedMessage
}
homeMonitor.eventHandlers["SignedForUserRequest"] = handlers.HandleSignedForUserRequest
homeMonitor.eventHandlers["SignedForAffirmation"] = handlers.HandleSignedForAffirmation
homeMonitor.eventHandlers["AffirmationCompleted"] = handlers.HandleAffirmationCompleted
homeMonitor.eventHandlers["AffirmationCompleted0"] = handlers.HandleAffirmationCompleted
homeMonitor.eventHandlers["CollectedSignatures"] = handlers.HandleCollectedSignatures
homeMonitor.eventHandlers["UserRequestForInformation"] = handlers.HandleUserRequestForInformation
homeMonitor.eventHandlers["SignedForInformation"] = handlers.HandleSignedForInformation
homeMonitor.eventHandlers["InformationRetrieved"] = handlers.HandleInformationRetrieved
homeMonitor.eventHandlers["ValidatorAdded"] = handlers.HandleValidatorAdded
homeMonitor.eventHandlers["ValidatorRemoved"] = handlers.HandleValidatorRemoved
foreignMonitor.eventHandlers["UserRequestForAffirmation"] = handlers.HandleUserRequestForAffirmation
foreignMonitor.eventHandlers["UserRequestForAffirmation0"] = handlers.HandleLegacyUserRequestForAffirmation
foreignMonitor.eventHandlers["RelayedMessage"] = handlers.HandleRelayedMessage
foreignMonitor.eventHandlers["RelayedMessage0"] = handlers.HandleRelayedMessage
foreignMonitor.eventHandlers["ValidatorAdded"] = handlers.HandleValidatorAdded
foreignMonitor.eventHandlers["ValidatorRemoved"] = handlers.HandleValidatorRemoved
return &Monitor{
@ -196,7 +211,13 @@ func (m *ContractMonitor) LoadUnprocessedLogs(ctx context.Context, fromBlock, to
var logs []*entity.Log
for {
var err error
logs, err = m.repo.Logs.FindByBlockRange(ctx, m.client.ChainID, m.cfg.Address, fromBlock, toBlock)
addresses := []common.Address{m.cfg.Address, m.cfg.ValidatorContractAddress}
if m.bridgeCfg.IsErcToNative {
for _, token := range m.bridgeCfg.Foreign.ErcToNativeTokens {
addresses = append(addresses, token.Address)
}
}
logs, err = m.repo.Logs.FindByBlockRange(ctx, m.client.ChainID, addresses, fromBlock, toBlock)
if err != nil {
m.logger.WithError(err).Error("can't find unprocessed logs in block range")
if utils.ContextSleep(ctx, 10*time.Second) == nil {
@ -213,6 +234,10 @@ func (m *ContractMonitor) LoadUnprocessedLogs(ctx context.Context, fromBlock, to
func (m *ContractMonitor) StartBlockFetcher(ctx context.Context, start uint) {
m.logger.Info("starting new blocks tracker")
if len(m.cfg.RefetchEvents) > 0 {
m.RefetchEvents(start - 1)
}
for {
head, err := m.client.BlockNumber(ctx)
if err != nil {
@ -229,11 +254,6 @@ func (m *ContractMonitor) StartBlockFetcher(ctx context.Context, start uint) {
}
m.headBlockMetric.Set(float64(m.headBlock))
if len(m.cfg.RefetchEvents) > 0 {
m.RefetchEvents(start - 1)
m.cfg.RefetchEvents = nil
}
for start <= m.headBlock {
end := start + m.cfg.MaxBlockRangeSize - 1
if end > m.headBlock {
@ -278,7 +298,7 @@ func (m *ContractMonitor) RefetchEvents(lastFetchedBlock uint) {
}
m.logger.WithFields(logrus.Fields{
"from_block": fromBlock,
"to_block": toBlock,
"to_block": end,
}).Info("scheduling new block range logs search")
m.blocksRangeChan <- &BlocksRange{
From: fromBlock,
@ -315,7 +335,8 @@ func (m *ContractMonitor) StartLogsFetcher(ctx context.Context) {
}
}
func (m *ContractMonitor) tryToFetchLogs(ctx context.Context, blocksRange *BlocksRange) error {
func (m *ContractMonitor) buildFilterQueries(blocksRange *BlocksRange) []ethereum.FilterQuery {
var qs []ethereum.FilterQuery
q := ethereum.FilterQuery{
FromBlock: big.NewInt(int64(blocksRange.From)),
ToBlock: big.NewInt(int64(blocksRange.To)),
@ -324,15 +345,48 @@ func (m *ContractMonitor) tryToFetchLogs(ctx context.Context, blocksRange *Block
if blocksRange.Topic != nil {
q.Topics = [][]common.Hash{{*blocksRange.Topic}}
}
var logs []types.Log
var err error
if m.cfg.Chain.SafeLogsRequest {
logs, err = m.client.FilterLogsSafe(ctx, q)
} else {
logs, err = m.client.FilterLogs(ctx, q)
qs = append(qs, q)
if m.bridgeCfg.IsErcToNative {
for _, token := range m.cfg.ErcToNativeTokens {
if token.StartBlock > 0 && blocksRange.To < token.StartBlock {
continue
}
if token.EndBlock > 0 && blocksRange.From > token.EndBlock {
continue
}
qc := q
if blocksRange.Topic != nil {
qc.Topics = [][]common.Hash{{*blocksRange.Topic}, {}, {m.cfg.Address.Hash()}}
} else {
qc.Topics = [][]common.Hash{{}, {}, {m.cfg.Address.Hash()}}
}
qc.Addresses = []common.Address{token.Address}
if token.StartBlock > 0 && token.StartBlock > blocksRange.From {
qc.FromBlock = big.NewInt(int64(token.StartBlock))
}
if token.EndBlock > 0 && token.EndBlock < blocksRange.To {
qc.ToBlock = big.NewInt(int64(token.EndBlock))
}
qs = append(qs, qc)
}
}
if err != nil {
return err
return qs
}
func (m *ContractMonitor) tryToFetchLogs(ctx context.Context, blocksRange *BlocksRange) error {
qs := m.buildFilterQueries(blocksRange)
var logs, logsBatch []types.Log
var err error
for _, q := range qs {
if m.cfg.Chain.SafeLogsRequest {
logsBatch, err = m.client.FilterLogsSafe(ctx, q)
} else {
logsBatch, err = m.client.FilterLogs(ctx, q)
}
if err != nil {
return err
}
logs = append(logs, logsBatch...)
}
m.logger.WithFields(logrus.Fields{
"count": len(logs),

73
presenter/presenter.go
View File

@ -195,16 +195,31 @@ func (p *Presenter) searchSentMessage(ctx context.Context, log *entity.Log) (*Se
return nil, nil
}
var messageInfo interface{}
var events []*EventInfo
msg, err := p.repo.Messages.FindByMsgHash(ctx, sent.BridgeID, sent.MsgHash)
if err != nil || msg == nil {
if err != nil {
return nil, err
}
events, err := p.buildMessageEvents(ctx, msg)
if msg == nil {
ercToNativeMsg, err2 := p.repo.ErcToNativeMessages.FindByMsgHash(ctx, sent.BridgeID, sent.MsgHash)
if err2 != nil {
return nil, err2
}
if ercToNativeMsg == nil {
return nil, nil
}
messageInfo = ercToNativeMessageToInfo(ercToNativeMsg)
events, err = p.buildMessageEvents(ctx, ercToNativeMsg.BridgeID, ercToNativeMsg.MsgHash, ercToNativeMsg.MsgHash)
} else {
messageInfo = messageToInfo(msg)
events, err = p.buildMessageEvents(ctx, msg.BridgeID, msg.MsgHash, msg.MessageID)
}
if err != nil {
return nil, err
}
return &SearchResult{
Message: messageToInfo(msg),
Message: messageInfo,
RelatedEvents: events,
}, nil
}
@ -218,16 +233,31 @@ func (p *Presenter) searchSignedMessage(ctx context.Context, log *entity.Log) (*
return nil, nil
}
var messageInfo interface{}
var events []*EventInfo
msg, err := p.repo.Messages.FindByMsgHash(ctx, sig.BridgeID, sig.MsgHash)
if err != nil || msg == nil {
if err != nil {
return nil, err
}
events, err := p.buildMessageEvents(ctx, msg)
if msg == nil {
ercToNativeMsg, err2 := p.repo.ErcToNativeMessages.FindByMsgHash(ctx, sig.BridgeID, sig.MsgHash)
if err2 != nil {
return nil, err2
}
if ercToNativeMsg == nil {
return nil, nil
}
messageInfo = ercToNativeMessageToInfo(ercToNativeMsg)
events, err = p.buildMessageEvents(ctx, ercToNativeMsg.BridgeID, ercToNativeMsg.MsgHash, ercToNativeMsg.MsgHash)
} else {
messageInfo = messageToInfo(msg)
events, err = p.buildMessageEvents(ctx, msg.BridgeID, msg.MsgHash, msg.MessageID)
}
if err != nil {
return nil, err
}
return &SearchResult{
Message: messageToInfo(msg),
Message: messageInfo,
RelatedEvents: events,
}, nil
}
@ -241,16 +271,31 @@ func (p *Presenter) searchExecutedMessage(ctx context.Context, log *entity.Log)
return nil, nil
}
var messageInfo interface{}
var events []*EventInfo
msg, err := p.repo.Messages.FindByMessageID(ctx, executed.BridgeID, executed.MessageID)
if err != nil || msg == nil {
if err != nil {
return nil, err
}
events, err := p.buildMessageEvents(ctx, msg)
if msg == nil {
ercToNativeMsg, err2 := p.repo.ErcToNativeMessages.FindByMsgHash(ctx, executed.BridgeID, executed.MessageID)
if err2 != nil {
return nil, err2
}
if ercToNativeMsg == nil {
return nil, nil
}
messageInfo = ercToNativeMessageToInfo(ercToNativeMsg)
events, err = p.buildMessageEvents(ctx, ercToNativeMsg.BridgeID, ercToNativeMsg.MsgHash, ercToNativeMsg.MsgHash)
} else {
messageInfo = messageToInfo(msg)
events, err = p.buildMessageEvents(ctx, msg.BridgeID, msg.MsgHash, msg.MessageID)
}
if err != nil {
return nil, err
}
return &SearchResult{
Message: messageToInfo(msg),
Message: messageInfo,
RelatedEvents: events,
}, nil
}
@ -324,20 +369,20 @@ func (p *Presenter) searchExecutedInformationRequest(ctx context.Context, log *e
}, nil
}
func (p *Presenter) buildMessageEvents(ctx context.Context, msg *entity.Message) ([]*EventInfo, error) {
sent, err := p.repo.SentMessages.FindByMsgHash(ctx, msg.BridgeID, msg.MsgHash)
func (p *Presenter) buildMessageEvents(ctx context.Context, bridgeID string, msgHash, messageID common.Hash) ([]*EventInfo, error) {
sent, err := p.repo.SentMessages.FindByMsgHash(ctx, bridgeID, msgHash)
if err != nil {
return nil, err
}
signed, err := p.repo.SignedMessages.FindByMsgHash(ctx, msg.BridgeID, msg.MsgHash)
signed, err := p.repo.SignedMessages.FindByMsgHash(ctx, bridgeID, msgHash)
if err != nil {
return nil, err
}
collected, err := p.repo.CollectedMessages.FindByMsgHash(ctx, msg.BridgeID, msg.MsgHash)
collected, err := p.repo.CollectedMessages.FindByMsgHash(ctx, bridgeID, msgHash)
if err != nil {
return nil, err
}
executed, err := p.repo.ExecutedMessages.FindByMessageID(ctx, msg.BridgeID, msg.MessageID)
executed, err := p.repo.ExecutedMessages.FindByMessageID(ctx, bridgeID, messageID)
if err != nil {
return nil, err
}

8
presenter/types.go
View File

@ -26,6 +26,14 @@ type InformationRequestInfo struct {
Executor common.Address
}
type ErcToNativeMessageInfo struct {
BridgeID string
MsgHash common.Hash
Direction entity.Direction
Receiver common.Address
Value string
}
type TxInfo struct {
BlockNumber uint
Timestamp time.Time

10
presenter/utils.go
View File

@ -43,3 +43,13 @@ func informationRequestToInfo(req *entity.InformationRequest) *InformationReques
Executor: req.Executor,
}
}
func ercToNativeMessageToInfo(req *entity.ErcToNativeMessage) *ErcToNativeMessageInfo {
return &ErcToNativeMessageInfo{
BridgeID: req.BridgeID,
MsgHash: req.MsgHash,
Direction: req.Direction,
Receiver: req.Receiver,
Value: req.Value,
}
}

44
prometheus/alertmanager.yml
View File

@ -97,6 +97,36 @@ receivers:
- type: button
text: 'Silence :no_bell:'
url: '{{ template "__alert_silence_link" . }}'
- name: slack-stuck-erc-to-native-message
slack_configs:
- send_resolved: true
channel: '#amb-alerts'
title: '{{ template "slack.stuck_erc_to_native_message.title" . }}'
text: '{{ template "slack.stuck_erc_to_native_message.text" . }}'
actions:
- type: button
text: 'Silence :no_bell:'
url: '{{ template "__alert_silence_link" . }}'
- name: slack-unknown-erc-to-native-confirmation
slack_configs:
- send_resolved: true
channel: '#amb-alerts'
title: '{{ template "slack.unknown_erc_to_native_confirmation.title" . }}'
text: '{{ template "slack.unknown_erc_to_native_confirmation.text" . }}'
actions:
- type: button
text: 'Silence :no_bell:'
url: '{{ template "__alert_silence_link" . }}'
- name: slack-unknown-erc-to-native-execution
slack_configs:
- send_resolved: true
channel: '#amb-alerts'
title: '{{ template "slack.unknown_erc_to_native_execution.title" . }}'
text: '{{ template "slack.unknown_erc_to_native_execution.text" . }}'
actions:
- type: button
text: 'Silence :no_bell:'
url: '{{ template "__alert_silence_link" . }}'
- name: slack-stuck-contract
slack_configs:
- send_resolved: true
@ -138,7 +168,7 @@ route:
matchers:
- alertname = FailedMessageExecution
- receiver: slack-stuck-information-request
group_by: [ "bridge_id", "chain_id", "block_number", "tx_hash" ]
group_by: [ "alertname", "bridge_id", "chain_id", "block_number", "tx_hash" ]
matchers:
- alertname = StuckInformationRequest
- receiver: slack-unknown-information-signature
@ -158,6 +188,18 @@ route:
group_by: [ "..." ]
matchers:
- alertname = DifferentInformationSignatures
- receiver: slack-stuck-erc-to-native-message
group_by: [ "alertname", "bridge_id", "chain_id", "block_number", "tx_hash", "receiver", "value" ]
matchers:
- alertname = StuckErcToNativeMessage
- receiver: slack-unknown-erc-to-native-confirmation
group_by: [ "..." ]
matchers:
- alertname = UnknownErcToNativeMessageConfirmation
- receiver: slack-unknown-erc-to-native-execution
group_by: [ "..." ]
matchers:
- alertname = UnknownErcToNativeMessageExecution
- receiver: slack-stuck-contract
group_by: [ "..." ]
matchers:

18
prometheus/rules.yml
View File

@ -53,6 +53,24 @@ groups:
expr: max_over_time(alert_monitor_different_information_signatures[5m]) > 0
annotations:
age: '{{ humanizeDuration $value }}'
- name: StuckErcToNativeMessage
rules:
- alert: StuckErcToNativeMessage
expr: max_over_time(alert_monitor_stuck_erc_to_native_message_confirmation[5m]) > 3600
annotations:
age: '{{ humanizeDuration $value }}'
- name: UnknownErcToNativeMessageConfirmation
rules:
- alert: UnknownErcToNativeMessageConfirmation
expr: max_over_time(alert_monitor_unknown_erc_to_native_message_confirmation[5m]) > 0
annotations:
age: '{{ humanizeDuration $value }}'
- name: UnknownErcToNativeMessageExecution
rules:
- alert: UnknownErcToNativeMessageExecution
expr: max_over_time(alert_monitor_unknown_erc_to_native_message_execution[5m]) > 0
annotations:
age: '{{ humanizeDuration $value }}'
- name: StuckContractProgress
rules:
- alert: StuckContractProgress

39
prometheus/templates/slack.tmpl
View File

@ -115,6 +115,45 @@ Validators signed different AMB information request results
*Tx:* {{ template "explorer.tx.link" .CommonLabels }}
{{- end }}
{{ define "slack.stuck_erc_to_native_message.title" -}}
Stuck ERC_TO_NATIVE message confirmation
{{- end }}
{{ define "slack.stuck_erc_to_native_message.text" -}}
*Bridge:* {{ .CommonLabels.bridge_id }}
*Chain ID:* {{ .CommonLabels.chain_id }}
*Block number:* {{ .CommonLabels.block_number }}
*Age:* {{ .CommonAnnotations.age }}
*Collected confirmations:* {{ .CommonLabels.count }}
*Receiver:* {{ .CommonLabels.receiver }}
*Value:* {{ .CommonLabels.value }}
*Tx:* {{ template "explorer.tx.link" .CommonLabels }}
{{- end }}
{{ define "slack.unknown_erc_to_native_confirmation.title" -}}
Validator signed for unknown ERC_TO_NATIVE message
{{- end }}
{{ define "slack.unknown_erc_to_native_confirmation.text" -}}
*Bridge:* {{ .CommonLabels.bridge_id }}
*Chain ID:* {{ .CommonLabels.chain_id }}
*Block number:* {{ .CommonLabels.block_number }}
*Age:* {{ .CommonAnnotations.age }}
*Validator:* {{ .CommonLabels.signer }}
*Message hash:* {{ .CommonLabels.msg_hash }}
*Tx:* {{ template "explorer.tx.link" .CommonLabels }}
{{- end }}
{{ define "slack.unknown_erc_to_native_execution.title" -}}
Bridge executed unknown ERC_TO_NATIVE message
{{- end }}
{{ define "slack.unknown_erc_to_native_execution.text" -}}
*Bridge:* {{ .CommonLabels.bridge_id }}
*Chain ID:* {{ .CommonLabels.chain_id }}
*Block number:* {{ .CommonLabels.block_number }}
*Age:* {{ .CommonAnnotations.age }}
*Message id:* {{ .CommonLabels.message_id }}
*Tx:* {{ template "explorer.tx.link" .CommonLabels }}
{{- end }}
{{ define "slack.stuck_contract.title" -}}
Monitoring of contract is stuck
{{- end }}

56
repository/postgres/erc_to_native_messages.go Normal file
View File

@ -0,0 +1,56 @@
package postgres
import (
"amb-monitor/db"
"amb-monitor/entity"
"context"
"database/sql"
"errors"
"fmt"
sq "github.com/Masterminds/squirrel"
"github.com/ethereum/go-ethereum/common"
)
type ercToNativeMessagesRepo basePostgresRepo
func NewErcToNativeMessagesRepo(table string, db *db.DB) entity.ErcToNativeMessagesRepo {
return (*ercToNativeMessagesRepo)(newBasePostgresRepo(table, db))
}
func (r *ercToNativeMessagesRepo) Ensure(ctx context.Context, msg *entity.ErcToNativeMessage) error {
q, args, err := sq.Insert(r.table).
Columns("bridge_id", "msg_hash", "direction", "receiver", "value").
Values(msg.BridgeID, msg.MsgHash, msg.Direction, msg.Receiver, msg.Value).
Suffix("ON CONFLICT (bridge_id, msg_hash) DO UPDATE SET updated_at = NOW()").
PlaceholderFormat(sq.Dollar).
ToSql()
if err != nil {
return fmt.Errorf("can't build query: %w", err)
}
_, err = r.db.ExecContext(ctx, q, args...)
if err != nil {
return fmt.Errorf("can't insert message: %w", err)
}
return nil
}
func (r *ercToNativeMessagesRepo) FindByMsgHash(ctx context.Context, bridgeID string, msgHash common.Hash) (*entity.ErcToNativeMessage, error) {
q, args, err := sq.Select("*").
From(r.table).
Where(sq.Eq{"bridge_id": bridgeID, "msg_hash": msgHash}).
PlaceholderFormat(sq.Dollar).
ToSql()
if err != nil {
return nil, fmt.Errorf("can't build query: %w", err)
}
msg := new(entity.ErcToNativeMessage)
err = r.db.GetContext(ctx, msg, q, args...)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return nil, fmt.Errorf("can't get message: %w", err)
}
return msg, nil
}

2
repository/postgres/logs.go
View File

@ -61,7 +61,7 @@ func (r *logsRepo) GetByID(ctx context.Context, id uint) (*entity.Log, error) {
return log, nil
}
func (r *logsRepo) FindByBlockRange(ctx context.Context, chainID string, addr common.Address, fromBlock uint, toBlock uint) ([]*entity.Log, error) {
func (r *logsRepo) FindByBlockRange(ctx context.Context, chainID string, addr []common.Address, fromBlock uint, toBlock uint) ([]*entity.Log, error) {
q, args, err := sq.Select("*").
From(r.table).
Where(sq.Eq{"chain_id": chainID, "address": addr}).

2
repository/repos.go
View File

@ -11,6 +11,7 @@ type Repo struct {
Logs entity.LogsRepo
BlockTimestamps entity.BlockTimestampsRepo
Messages entity.MessagesRepo
ErcToNativeMessages entity.ErcToNativeMessagesRepo
SentMessages entity.SentMessagesRepo
SignedMessages entity.SignedMessagesRepo
CollectedMessages entity.CollectedMessagesRepo
@ -28,6 +29,7 @@ func NewRepo(db *db.DB) *Repo {
Logs: postgres.NewLogsRepo("logs", db),
BlockTimestamps: postgres.NewBlockTimestampsRepo("block_timestamps", db),
Messages: postgres.NewMessagesRepo("messages", db),
ErcToNativeMessages: postgres.NewErcToNativeMessagesRepo("erc_to_native_messages", db),
SentMessages: postgres.NewSentMessagesRepo("sent_messages", db),
SignedMessages: postgres.NewSignedMessagesRepo("signed_messages", db),
CollectedMessages: postgres.NewCollectedMessagesRepo("collected_messages", db),