|
| 1 | +// Copyright (c) 2026 IOTA Stiftung |
| 2 | +// SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +import { setTimeout as sleep } from 'node:timers/promises'; |
| 5 | +import type { GrpcNetwork } from '@iota/grpc'; |
| 6 | +import { CheckpointResponseField, GRPC_URLS, IotaGrpcClient } from '@iota/grpc'; |
| 7 | + |
| 8 | +/** |
| 9 | + * Example: Checkpoint Follower |
| 10 | + * |
| 11 | + * This example follows a network over gRPC, printing each checkpoint as the |
| 12 | + * network produces it. When the stream drops, it starts a new one after the |
| 13 | + * last checkpoint it received, so none are skipped. Press Ctrl+C to stop. |
| 14 | + * |
| 15 | + * It follows the network given as its first argument (mainnet, testnet, devnet |
| 16 | + * or localnet), or testnet without one. It starts at the latest checkpoint, or |
| 17 | + * at the sequence number given as its second argument: |
| 18 | + * |
| 19 | + * pnpm example ./src/grpc-checkpoint-follower.ts |
| 20 | + * pnpm example ./src/grpc-checkpoint-follower.ts mainnet |
| 21 | + * pnpm example ./src/grpc-checkpoint-follower.ts testnet 274000000 |
| 22 | + */ |
| 23 | + |
| 24 | +// Give up after this many failures in a row without receiving a checkpoint. |
| 25 | +const MAX_FAILURES = 5; |
| 26 | + |
| 27 | +// Read the network from the first argument, defaulting to testnet |
| 28 | +const network = process.argv[2] || 'testnet'; |
| 29 | +if (!Object.hasOwn(GRPC_URLS, network)) { |
| 30 | + throw new Error( |
| 31 | + `Unknown network '${network}': use one of ${Object.keys(GRPC_URLS).join(', ')}`, |
| 32 | + ); |
| 33 | +} |
| 34 | + |
| 35 | +// Create a client connected to that network |
| 36 | +const client = new IotaGrpcClient({ network: network as GrpcNetwork }); |
| 37 | + |
| 38 | +// Stop on Ctrl+C |
| 39 | +const controller = new AbortController(); |
| 40 | +process.on('SIGINT', () => controller.abort()); |
| 41 | + |
| 42 | +// The next checkpoint to ask for. Undefined asks for the latest. |
| 43 | +let next = process.argv[3] === undefined ? undefined : BigInt(process.argv[3]); |
| 44 | +let failures = 0; |
| 45 | + |
| 46 | +console.log(`Following ${network} from ${next ?? 'the latest checkpoint'}...`); |
| 47 | + |
| 48 | +while (!controller.signal.aborted) { |
| 49 | + try { |
| 50 | + const stream = client.streamCheckpoints({ |
| 51 | + startSequenceNumber: next, |
| 52 | + // By default only the summary is sent: ask for the transaction digests too |
| 53 | + readMask: [ |
| 54 | + CheckpointResponseField.CHECKPOINT_SUMMARY, |
| 55 | + CheckpointResponseField.TRANSACTIONS_TRANSACTION_DIGEST, |
| 56 | + ], |
| 57 | + signal: controller.signal, |
| 58 | + }); |
| 59 | + |
| 60 | + for await (const item of stream.items) { |
| 61 | + if (item.kind !== 'checkpoint') { |
| 62 | + continue; |
| 63 | + } |
| 64 | + |
| 65 | + console.log( |
| 66 | + `Checkpoint ${item.sequenceNumber}: ${item.transactions.length} transactions`, |
| 67 | + ); |
| 68 | + next = item.sequenceNumber + 1n; |
| 69 | + failures = 0; |
| 70 | + } |
| 71 | + } catch (error) { |
| 72 | + // Ctrl+C ends the stream with a CANCELED error |
| 73 | + if (controller.signal.aborted) { |
| 74 | + break; |
| 75 | + } |
| 76 | + |
| 77 | + failures++; |
| 78 | + if (failures > MAX_FAILURES) { |
| 79 | + throw error; |
| 80 | + } |
| 81 | + |
| 82 | + // Wait 1, 2, 4, 8 and 16 seconds between attempts |
| 83 | + const delayMs = 1_000 * 2 ** (failures - 1); |
| 84 | + console.warn(`Stream dropped: ${(error as Error).message}`); |
| 85 | + console.warn(`Resuming at ${next ?? 'the latest checkpoint'} in ${delayMs / 1_000}s...`); |
| 86 | + await sleep(delayMs, undefined, { signal: controller.signal }).catch(() => {}); |
| 87 | + } |
| 88 | +} |
| 89 | + |
| 90 | +console.log('Stopped.'); |
0 commit comments