Skip to content

Commit 764d7d9

Browse files
panoelevan-gray
panoel
authored and
evan-gray
committed
watcher: Use signedVAAs instead of v2Events table, self-healing feature
1 parent 94b5ceb commit 764d7d9

14 files changed

+160
-82
lines changed

‎watcher/.env.sample

+1-1
Original file line numberDiff line numberDiff line change
@@ -10,5 +10,5 @@ FIRESTORE_LATEST_COLLECTION=
1010
GOOGLE_APPLICATION_CREDENTIALS=
1111
BIGTABLE_TABLE_ID=
1212
BIGTABLE_INSTANCE_ID=
13-
BIGTABLE_VAA_TABLE_ID=
13+
BIGTABLE_SIGNED_VAAS_TABLE_ID=
1414
BIGTABLE_VAAS_BY_TX_HASH_TABLE_ID=

‎watcher/package.json

+1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
"backfill-vaas-by-tx-hash": "ts-node scripts/backfillVAAsByTxHash.ts",
1616
"locate-message-gaps": "ts-node scripts/locateMessageGaps.ts",
1717
"fetch-missing-vaas": "ts-node scripts/fetchMissingVAAs.ts",
18+
"update-found-vaas": "ts-node scripts/updateFoundVAAs.ts",
1819
"read-bigtable": "ts-node scripts/readBigtable.ts",
1920
"read-firestore": "ts-node scripts/readFirestore.ts"
2021
},

‎watcher/scripts/backfillVAAsByTxHash.ts

+4-5
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import {
88
makeVAAsByTxHashRowKey,
99
parseMessageId,
1010
} from '../src/databases/utils';
11-
import { ChainId } from '@certusone/wormhole-sdk';
1211
import { chunkArray } from '@wormhole-foundation/wormhole-monitor-common';
1312

1413
const CHUNK_SIZE = 10000;
@@ -20,17 +19,17 @@ const CHUNK_SIZE = 10000;
2019
throw new Error('bigtable is undefined');
2120
}
2221
const instance = bt.bigtable.instance(bt.instanceId);
23-
const messageTable = instance.table(bt.tableId);
22+
const messageTable = instance.table(bt.msgTableId);
2423
const vaasByTxHashTable = instance.table(bt.vaasByTxHashTableId);
2524

26-
let log = ora(`Reading rows from ${bt.tableId}...`).start();
25+
let log = ora(`Reading rows from ${bt.msgTableId}...`).start();
2726
const observedMessages = await messageTable.getRows(); // TODO: pagination
2827
const vaasByTxHash: { [key: string]: string[] } = {};
2928
for (const msg of observedMessages[0]) {
3029
const txHash = msg.data.info.txHash[0].value;
3130
const { chain, emitter, sequence } = parseMessageId(msg.id);
32-
const txHashRowKey = makeVAAsByTxHashRowKey(txHash, chain as ChainId);
33-
const vaaRowKey = makeSignedVAAsRowKey(chain as ChainId, emitter, sequence.toString());
31+
const txHashRowKey = makeVAAsByTxHashRowKey(txHash, chain);
32+
const vaaRowKey = makeSignedVAAsRowKey(chain, emitter, sequence.toString());
3433
vaasByTxHash[txHashRowKey] = [...(vaasByTxHash[txHashRowKey] || []), vaaRowKey];
3534
}
3635
const rowsToInsert = Object.entries(vaasByTxHash).map<BigtableVAAsByTxHashRow>(

‎watcher/scripts/deleteMessagesByChain.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ const CHAIN = CHAIN_ID_SOLANA;
1616
}
1717

1818
const instance = bt.bigtable.instance(bt.instanceId);
19-
const messageTable = instance.table(bt.tableId);
19+
const messageTable = instance.table(bt.msgTableId);
2020
await messageTable.deleteRows(`${padUint16(coalesceChainId(CHAIN).toString())}/`);
2121
console.log('Deleted all rows starting with', coalesceChainName(CHAIN));
2222
})();

‎watcher/scripts/fetchMissingVAAs.ts

+6-14
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,8 @@ import axios from 'axios';
44
import { writeFileSync } from 'fs';
55
import ora from 'ora';
66
import { BigtableDatabase } from '../src/databases/BigtableDatabase';
7-
import { makeVaaId, parseMessageId } from '../src/databases/utils';
8-
import { AXIOS_CONFIG_JSON } from '../src/consts';
7+
import { makeSignedVAAsRowKey, parseMessageId } from '../src/databases/utils';
8+
import { AXIOS_CONFIG_JSON, GUARDIAN_RPC_HOSTS } from '../src/consts';
99
import { parseVaa } from '@certusone/wormhole-sdk';
1010

1111
// This script checks for messages which don't have VAAs and attempts to fetch the VAAs from the guardians
@@ -17,14 +17,6 @@ import { parseVaa } from '@certusone/wormhole-sdk';
1717
const foundVaas: { [id: string]: string } = {};
1818
const missingVaas: { [id: string]: string | undefined } = {};
1919

20-
const GUARDIAN_RPCS = [
21-
'https://wormhole-v2-mainnet-api.certus.one',
22-
'https://wormhole.inotel.ro',
23-
'https://wormhole-v2-mainnet-api.mcf.rocks',
24-
'https://wormhole-v2-mainnet-api.chainlayer.network',
25-
'https://wormhole-v2-mainnet-api.staking.fund',
26-
];
27-
2820
(async () => {
2921
const bt = new BigtableDatabase();
3022
if (!bt.bigtable) {
@@ -43,13 +35,13 @@ const GUARDIAN_RPCS = [
4335
for (const observedMessage of missingVaaMessages) {
4436
log.text = `Searching for VAA ${++search}/${total}...`;
4537
const { chain, emitter, sequence } = parseMessageId(observedMessage.id);
46-
const id = makeVaaId(chain, emitter, sequence);
38+
const id = makeSignedVAAsRowKey(chain, emitter, sequence.toString());
4739
let vaaBytes: string | null = null;
48-
for (const rpc of GUARDIAN_RPCS) {
49-
log.text = `Searching for VAA ${search}/${total} (${rpc})...`;
40+
for (const host of GUARDIAN_RPC_HOSTS) {
41+
log.text = `Searching for VAA ${search}/${total} (${host})...`;
5042
try {
5143
const result = await axios.get(
52-
`${rpc}/v1/signed_vaa/${chain}/${emitter}/${sequence.toString()}`,
44+
`${host}/v1/signed_vaa/${chain}/${emitter}/${sequence.toString()}`,
5345
AXIOS_CONFIG_JSON
5446
);
5547
if (result.data.vaaBytes) {

‎watcher/scripts/locateMessageGaps.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ import { Watcher } from '../src/watchers/Watcher';
1717
throw new Error('bigtable is undefined');
1818
}
1919
const instance = bt.bigtable.instance(bt.instanceId);
20-
const messageTable = instance.table(bt.tableId);
20+
const messageTable = instance.table(bt.msgTableId);
2121
try {
2222
// Find gaps in sequence numbers with the same chain and emitter
2323
// Sort by ascending sequence number

‎watcher/scripts/readBigtable.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import { BigtableDatabase } from '../src/databases/BigtableDatabase';
1212
throw new Error('bigtable is undefined');
1313
}
1414
const mainnetInstance = bt.bigtable.instance(bt.instanceId);
15-
const messageTable = mainnetInstance.table(bt.tableId);
15+
const messageTable = mainnetInstance.table(bt.msgTableId);
1616
try {
1717
const chain: ChainId = 22;
1818
const prefix = `${padUint16(chain.toString())}/`;

‎watcher/scripts/updateFoundVAAs.ts

+30
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
import * as dotenv from 'dotenv';
2+
dotenv.config();
3+
import { BigtableDatabase } from '../src/databases/BigtableDatabase';
4+
5+
// This script takes the output of fetchMissingVAAs and writes the found records back to the VAA big table
6+
7+
(async () => {
8+
const found: { [id: string]: string } = require('../found.json');
9+
const bt = new BigtableDatabase();
10+
if (!bt.bigtable) {
11+
throw new Error('bigtable is undefined');
12+
}
13+
try {
14+
bt.storeSignedVAAs(
15+
Object.entries(found).map(([id, vaaBytes]) => {
16+
const vaa = Buffer.from(vaaBytes, 'hex');
17+
return {
18+
key: id,
19+
data: {
20+
info: {
21+
bytes: { value: vaa, timestamp: '0' },
22+
},
23+
},
24+
};
25+
})
26+
);
27+
} catch (e) {
28+
console.error(e);
29+
}
30+
})();

‎watcher/scripts/updateRows.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ export function parseVaaId(vaaRowKey: string) {
1818
throw new Error('bigtable is undefined');
1919
}
2020
const instance = bt.bigtable.instance(bt.instanceId);
21-
const messageTable = instance.table(bt.tableId);
21+
const messageTable = instance.table(bt.msgTableId);
2222

2323
const rowKeysToUpdate: string[] = [
2424
'5:0000000000000000000000005a58505a96d1dbf8df91cb21b54419fc36e93fde:0000000000006840',

‎watcher/src/consts.ts

+8
Original file line numberDiff line numberDiff line change
@@ -74,3 +74,11 @@ export const DB_LAST_BLOCK_FILE = process.env.DB_LAST_BLOCK_FILE || './lastBlock
7474
export const AXIOS_CONFIG_JSON: AxiosRequestConfig = {
7575
headers: { 'Accept-Encoding': 'application/json' },
7676
};
77+
78+
export const GUARDIAN_RPC_HOSTS = [
79+
'https://wormhole-v2-mainnet-api.certus.one',
80+
'https://wormhole.inotel.ro',
81+
'https://wormhole-v2-mainnet-api.mcf.rocks',
82+
'https://wormhole-v2-mainnet-api.chainlayer.network',
83+
'https://wormhole-v2-mainnet-api.staking.fund',
84+
];

‎watcher/src/databases/BigtableDatabase.ts

+68-40
Original file line numberDiff line numberDiff line change
@@ -12,30 +12,33 @@ import { Database } from './Database';
1212
import {
1313
BigtableMessagesResultRow,
1414
BigtableMessagesRow,
15+
BigtableSignedVAAsResultRow,
16+
BigtableSignedVAAsRow,
1517
BigtableVAAsByTxHashRow,
16-
BigtableVAAsResultRow,
1718
VaasByBlock,
1819
} from './types';
1920
import {
2021
makeMessageId,
2122
makeVAAsByTxHashRowKey,
22-
makeVaaId,
2323
makeSignedVAAsRowKey,
2424
parseMessageId,
2525
} from './utils';
26+
import { getSignedVAA } from '../utils/getSignedVAA';
2627

2728
const WATCH_MISSING_TIMEOUT = 5 * 60 * 1000;
2829

2930
export class BigtableDatabase extends Database {
30-
tableId: string;
31+
msgTableId: string;
32+
signedVAAsTableId: string;
3133
vaasByTxHashTableId: string;
3234
instanceId: string;
3335
bigtable: Bigtable;
3436
firestoreDb: FirebaseFirestore.Firestore;
3537
latestCollectionName: string;
3638
constructor() {
3739
super();
38-
this.tableId = assertEnvironmentVariable('BIGTABLE_TABLE_ID');
40+
this.msgTableId = assertEnvironmentVariable('BIGTABLE_TABLE_ID');
41+
this.signedVAAsTableId = assertEnvironmentVariable('BIGTABLE_SIGNED_VAAS_TABLE_ID');
3942
this.vaasByTxHashTableId = assertEnvironmentVariable('BIGTABLE_VAAS_BY_TX_HASH_TABLE_ID');
4043
this.instanceId = assertEnvironmentVariable('BIGTABLE_INSTANCE_ID');
4144
this.latestCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_COLLECTION');
@@ -92,7 +95,7 @@ export class BigtableDatabase extends Database {
9295
const chainId = coalesceChainId(chain);
9396
const filteredBlocks = BigtableDatabase.filterEmptyBlocks(vaasByBlock);
9497
const instance = this.bigtable.instance(this.instanceId);
95-
const table = instance.table(this.tableId);
98+
const table = instance.table(this.msgTableId);
9699
const vaasByTxHashTable = instance.table(this.vaasByTxHashTableId);
97100
const rowsToInsert: BigtableMessagesRow[] = [];
98101
const vaasByTxHash: { [key: string]: string[] } = {};
@@ -122,7 +125,7 @@ export class BigtableDatabase extends Database {
122125
},
123126
},
124127
});
125-
const txHashRowKey = makeVAAsByTxHashRowKey(txHash, chain);
128+
const txHashRowKey = makeVAAsByTxHashRowKey(txHash, chainId);
126129
const vaaRowKey = makeSignedVAAsRowKey(chainId, emitter, seq);
127130
vaasByTxHash[txHashRowKey] = [...(vaasByTxHash[txHashRowKey] || []), vaaRowKey];
128131
});
@@ -154,7 +157,7 @@ export class BigtableDatabase extends Database {
154157

155158
async updateMessageStatuses(messageKeys: string[], value: number = 1): Promise<void> {
156159
const instance = this.bigtable.instance(this.instanceId);
157-
const table = instance.table(this.tableId);
160+
const table = instance.table(this.msgTableId);
158161
const chunkedMessageKeys = chunkArray(messageKeys, 1000);
159162
for (const chunk of chunkedMessageKeys) {
160163
const rowsToInsert: BigtableMessagesRow[] = chunk.map((id) => ({
@@ -175,7 +178,7 @@ export class BigtableDatabase extends Database {
175178

176179
async fetchMissingVaaMessages(): Promise<BigtableMessagesResultRow[]> {
177180
const instance = this.bigtable.instance(this.instanceId);
178-
const messageTable = instance.table(this.tableId);
181+
const messageTable = instance.table(this.msgTableId);
179182
// TODO: how to filter to only messages with hasSignedVaa === 0
180183
const observedMessages = (await messageTable.getRows())[0] as BigtableMessagesResultRow[];
181184
const missingVaaMessages = observedMessages.filter(
@@ -185,70 +188,95 @@ export class BigtableDatabase extends Database {
185188
}
186189

187190
async watchMissing(): Promise<void> {
188-
const vaaTableId = assertEnvironmentVariable('BIGTABLE_VAA_TABLE_ID');
189191
const instance = this.bigtable.instance(this.instanceId);
190-
const vaaTable = instance.table(vaaTableId);
192+
const signedVAAsTable = instance.table(this.signedVAAsTableId);
191193
while (true) {
192194
try {
195+
// this array first stores all of the messages which are missing VAAs
196+
// messages which we find VAAs for are then pruned from the array
197+
// lastly we try to fetch VAAs for the messages in the pruned array from the guardians
193198
const missingVaaMessages = await this.fetchMissingVaaMessages();
194199
const total = missingVaaMessages.length;
195200
this.logger.info(`locating ${total} messages with hasSignedVAA === 0`);
196201
let found = 0;
197202
const chunkedVAAIds = chunkArray(
198203
missingVaaMessages.map((observedMessage) => {
199204
const { chain, emitter, sequence } = parseMessageId(observedMessage.id);
200-
return makeVaaId(chain, emitter, sequence);
205+
return makeSignedVAAsRowKey(chain, emitter, sequence.toString());
201206
}),
202207
1000
203208
);
204209
let chunkNum = 0;
205-
const foundRecords: string[] = [];
210+
const foundKeys: string[] = [];
206211
for (const chunk of chunkedVAAIds) {
207212
this.logger.info(`processing chunk ${++chunkNum} of ${chunkedVAAIds.length}`);
208-
const filter = [
209-
{
210-
family: 'QuorumState',
211-
column: 'SignedVaa',
212-
},
213-
];
214213
const vaaRows = (
215-
await vaaTable.getRows({
214+
await signedVAAsTable.getRows({
216215
keys: chunk,
217216
decode: false,
218-
filter,
219217
})
220-
)[0] as BigtableVAAsResultRow[];
218+
)[0] as BigtableSignedVAAsResultRow[];
221219
for (const row of vaaRows) {
222220
try {
223-
const vaaBytes = row.data.QuorumState.SignedVAA?.[0].value;
224-
if (vaaBytes) {
225-
const parsed = parseVaa(vaaBytes);
226-
const matchingIndex = missingVaaMessages.findIndex((observedMessage) => {
227-
const { chain, emitter, sequence } = parseMessageId(observedMessage.id);
228-
if (
229-
parsed.emitterChain === chain &&
230-
parsed.emitterAddress.toString('hex') === emitter &&
231-
parsed.sequence === sequence
232-
) {
233-
return true;
234-
}
235-
});
236-
if (matchingIndex !== -1) {
237-
found++;
238-
// remove matches to keep array lean
239-
const [matching] = missingVaaMessages.splice(matchingIndex, 1);
240-
foundRecords.push(matching.id);
221+
const vaaBytes = row.data.info.bytes[0].value;
222+
const parsed = parseVaa(vaaBytes);
223+
const matchingIndex = missingVaaMessages.findIndex((observedMessage) => {
224+
const { chain, emitter, sequence } = parseMessageId(observedMessage.id);
225+
if (
226+
parsed.emitterChain === chain &&
227+
parsed.emitterAddress.toString('hex') === emitter &&
228+
parsed.sequence === sequence
229+
) {
230+
return true;
241231
}
232+
});
233+
if (matchingIndex !== -1) {
234+
found++;
235+
// remove matches to keep array lean
236+
// messages with missing VAAs will be kept in the array
237+
const [matching] = missingVaaMessages.splice(matchingIndex, 1);
238+
foundKeys.push(matching.id);
242239
}
243240
} catch (e) {}
244241
}
245242
}
246243
this.logger.info(`processed ${total} messages, found ${found}, missing ${total - found}`);
247-
this.updateMessageStatuses(foundRecords);
244+
this.updateMessageStatuses(foundKeys);
245+
// attempt to fetch VAAs missing from messages from the guardians and store them
246+
// this is useful for cases where the VAA doesn't exist in the `signedVAAsTable` (perhaps due to an outage) but is available
247+
const missingSignedVAARows: BigtableSignedVAAsRow[] = [];
248+
for (const msg of missingVaaMessages) {
249+
const { chain, emitter, sequence } = parseMessageId(msg.id);
250+
const seq = sequence.toString();
251+
const vaaBytes = await getSignedVAA(chain, emitter, seq);
252+
if (vaaBytes) {
253+
const key = makeSignedVAAsRowKey(chain, emitter, seq);
254+
missingSignedVAARows.push({
255+
key,
256+
data: {
257+
info: {
258+
bytes: { value: vaaBytes, timestamp: '0' },
259+
},
260+
},
261+
});
262+
}
263+
}
264+
this.storeSignedVAAs(missingSignedVAARows);
265+
// TODO: add slack message alerts
248266
} catch (e) {
249267
this.logger.error(e);
250268
}
251269
await sleep(WATCH_MISSING_TIMEOUT);
252270
}
253271
}
272+
273+
async storeSignedVAAs(rows: BigtableSignedVAAsRow[]): Promise<void> {
274+
const instance = this.bigtable.instance(this.instanceId);
275+
const table = instance.table(this.signedVAAsTableId);
276+
const chunks = chunkArray(rows, 1000);
277+
for (const chunk of chunks) {
278+
await table.insert(chunk);
279+
this.logger.info(`wrote ${chunk.length} signed VAAs to the ${this.signedVAAsTableId} table`);
280+
}
281+
}
254282
}

0 commit comments

Comments
 (0)