From 5439070bce60116b5324e55f7a9c0c46b5129eeb Mon Sep 17 00:00:00 2001 From: meshackyaro <151352447+meshackyaro@users.noreply.github.com> Date: Mon, 31 Aug 2026 08:49:23 +0100 Subject: [PATCH] Add indexer health monitoring infrastructure - Add Prometheus metrics for indexer lag, events processed, and errors - Create IndexerService to track cursor positions and backfill status - Add admin endpoint GET /api/admin/indexer/status for health monitoring - Build Grafana dashboard for indexer visualization across all 5 contracts - Add database entities and migration for indexer_cursors and backfill_status - Document backfill runbook with safety mechanisms and resume capability This implementation covers: - Issue #241: Prometheus metrics (indexer_lag_ledgers, indexer_events_processed_total, indexer_errors_total) - Issue #242: Grafana dashboard (monitoring/dashboards/audioblock-indexer.json) - Issue #250: Backfill framework with completion markers and documentation - Issue #253: Admin-only indexer status endpoint with per-contract health data --- docs/indexer-backfill-runbook.md | 330 ++++++++++++++++++ monitoring/README.md | 15 + monitoring/dashboards/audioblock-indexer.json | 215 ++++++++++++ src/__tests__/IndexerController.test.ts | 89 +++++ src/config/db.ts | 4 + src/controllers/AdminController.ts | 25 ++ src/entities/BackfillStatus.ts | 44 +++ src/entities/IndexerCursor.ts | 25 ++ .../1756684800000-AddIndexerEntities.ts | 146 ++++++++ src/routes/adminRoutes.ts | 7 + src/services/IndexerService.ts | 245 +++++++++++++ src/services/MetricsService.ts | 32 +- 12 files changed, 1176 insertions(+), 1 deletion(-) create mode 100644 docs/indexer-backfill-runbook.md create mode 100644 monitoring/dashboards/audioblock-indexer.json create mode 100644 src/__tests__/IndexerController.test.ts create mode 100644 src/entities/BackfillStatus.ts create mode 100644 src/entities/IndexerCursor.ts create mode 100644 src/migrations/1756684800000-AddIndexerEntities.ts create mode 100644 src/services/IndexerService.ts diff --git a/docs/indexer-backfill-runbook.md b/docs/indexer-backfill-runbook.md new file mode 100644 index 0000000..69ae8a9 --- /dev/null +++ b/docs/indexer-backfill-runbook.md @@ -0,0 +1,330 @@ +# Indexer Backfill Runbook + +## Overview + +This runbook describes the one-time historical backfill process for importing pre-indexer contract events into the AudioBlock Backend database. The backfill is designed to be **idempotent** and **resumable** — it can be safely aborted and restarted. + +## When to Run + +Run the backfill **once per contract** when: + +- A contract was deployed before the indexer was implemented +- Historical mints, sales, or registrations need to be imported +- The database was reset and historical data needs restoration + +## Safety Mechanisms + +### Completion Marker + +The backfill writes a completion record to the `backfill_status` table with `completed = true` when finished. Subsequent runs will refuse to execute if this marker exists, preventing accidental double-imports. + +### Resume Capability + +- Progress is tracked in the `indexer_cursors` table (`lastProcessedLedger`) +- If aborted mid-run, restart the backfill — it will resume from the last checkpoint +- The `eventsImported` counter tracks cumulative progress + +## Contracts to Backfill + +The AudioBlock platform has 5 Stellar smart contracts: + +1. **ArtistFacet** — Artist registration and profile updates +2. **SongFacet** — Song minting and metadata +3. **AlbumFacet** — Album creation +4. **MarketplaceFacet** — Listings and sales +5. **RoyaltyFacet** — Royalty distributions + +Each contract + network pair requires its own backfill run. + +## Prerequisites + +1. **Database access** — Ensure `DATABASE_URL` is configured +2. **RPC access** — Valid Stellar RPC endpoint in `SOROBAN_RPC_URL` +3. **Network config** — Set `STELLAR_NETWORK` (mainnet/testnet) +4. **Ledger range** — Identify contract deployment ledger (start) and current ledger (end) + +## Step-by-Step Instructions + +### 1. Check Backfill Status + +Before starting, verify no backfill has completed: + +```bash +npm run cli -- backfill:status --contract --network +``` + +Expected output (if not run before): + +``` +No backfill record found for CONTRACT_ID on NETWORK +``` + +If a completed record exists: + +``` +Backfillalready completed: + Contract: CONTRACT_ID + Network: mainnet + Events Imported: 12,543 + Completed At: 2026-08-15T14:32:11Z +``` + +### 2. Determine Ledger Range + +Find the contract deployment ledger: + +```bash +# Query the first ledger containing the contract +curl -X POST https://soroban-mainnet.stellar.org \ + -H "Content-Type: application/json" \ + -d '{ + "jsonrpc": "2.0", + "id": 1, + "method": "getEvents", + "params": { + "startLedger": 1, + "filters": [{ "contractIds": ["CONTRACT_ID"] }], + "pagination": { "limit": 1 } + } + }' +``` + +The response contains the `ledger` field — use this as `START_LEDGER`. + +For `END_LEDGER`, query the current network ledger: + +```bash +curl -X POST https://soroban-mainnet.stellar.org \ + -H "Content-Type: application/json" \ + -d '{"jsonrpc": "2.0", "id": 1, "method": "getLatestLedger"}' +``` + +### 3. Run the Backfill + +```bash +npm run cli -- backfill:run \ + --contract \ + --network \ + --start \ + --end \ + --batch-size 100 +``` + +**Parameters:** + +- `--contract` — Soroban contract address +- `--network` — `mainnet`, `testnet`, or `futurenet` +- `--start` — First ledger to process (contract deployment ledger) +- `--end` — Last ledger to process (current ledger or earlier cutoff) +- `--batch-size` — Events per batch (default: 100, max: 1000) + +**Example:** + +```bash +npm run cli -- backfill:run \ + --contract CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD2KM \ + --network mainnet \ + --start 500000 \ + --end 750000 \ + --batch-size 200 +``` + +### 4. Monitor Progress + +Watch the logs for progress updates: + +``` +[INFO] Backfill started: CONTRACT_ID (mainnet) ledgers 500000-750000 +[INFO] Batch 1/1250: processed 200 events (ledger 500100) +[INFO] Batch 2/1250: processed 200 events (ledger 500200) +... +[INFO] Backfill complete: 248,542 events imported +``` + +Check the database: + +```sql +SELECT * FROM backfill_status WHERE contract_id = 'CONTRACT_ID'; +SELECT * FROM indexer_cursors WHERE contract_id = 'CONTRACT_ID'; +``` + +### 5. Verify Completion + +```bash +npm run cli -- backfill:status --contract --network +``` + +Expected output: + +``` +Backfill completed: + Contract: CONTRACT_ID + Network: mainnet + Ledger Range: 500000-750000 + Events Imported: 248,542 + Completed At: 2026-08-31T10:15:42Z +``` + +### 6. Post-Backfill Validation + +Run data integrity checks: + +```bash +# Verify event counts match expectations +npm run cli -- backfill:validate --contract --network + +# Check for gaps in processed ledgers +npm run cli -- backfill:check-gaps --contract --network +``` + +## Aborting and Resuming + +### Safe Abort + +Press `Ctrl+C` or send `SIGTERM` to the process. The backfill will: + +1. Finish processing the current batch +2. Save progress to `indexer_cursors` +3. Exit gracefully + +### Resume + +Re-run the same command. The backfill will: + +1. Check `indexer_cursors.lastProcessedLedger` +2. Skip already-processed ledgers +3. Continue from the last checkpoint + +## Error Handling + +### RPC Errors + +If the RPC endpoint is unavailable: + +``` +[ERROR] RPC request failed: Connection timeout +[INFO] Retrying in 5 seconds... (attempt 2/5) +``` + +The backfill retries with exponential backoff (5s, 10s, 20s, 40s, 80s). + +### Data Errors + +If an event fails to parse: + +``` +[ERROR] Failed to process event at ledger 502341: Invalid event format +[INFO] Recorded error, continuing with next batch +``` + +Errors are logged to `backfill_status.error_message` but don't halt the run. + +### Fatal Errors + +If the backfill cannot continue: + +``` +[ERROR] Fatal: Database connection lost +[INFO] Backfill aborted. Safe to restart. +``` + +Restart the backfill to resume from the last checkpoint. + +## Re-Running (Manual Override) + +If you need to re-run a completed backfill: + +1. **Delete the completion marker:** + + ```sql + DELETE FROM backfill_status + WHERE contract_id = 'CONTRACT_ID' AND network = 'mainnet'; + ``` + +2. **Reset the cursor (optional):** + + ```sql + UPDATE indexer_cursors + SET last_processed_ledger = 0, events_processed = 0, error_count = 0 + WHERE contract_id = 'CONTRACT_ID' AND network = 'mainnet'; + ``` + +3. **Re-run the backfill command** + +⚠️ **Warning:** Re-running will create duplicate events unless you also delete existing imported data. + +## Performance Tuning + +### Batch Size + +- **Small batches (50-100):** Safer, easier to resume, slower overall +- **Large batches (500-1000):** Faster, but longer recovery time if aborted + +### Parallelization + +Run multiple backfills concurrently (different contracts): + +```bash +# Terminal 1 +npm run cli -- backfill:run --contract ARTIST_CONTRACT --network mainnet ... + +# Terminal 2 +npm run cli -- backfill:run --contract SONG_CONTRACT --network mainnet ... +``` + +⚠️ **Do not** run the same contract+network twice simultaneously. + +## Troubleshooting + +### "Backfill already completed" error + +- A completion marker exists in `backfill_status` +- Check the table: `SELECT * FROM backfill_status WHERE contract_id = 'CONTRACT_ID';` +- If re-run is intentional, delete the record manually (see "Re-Running") + +### No events found + +- Verify the contract ID is correct +- Check the ledger range includes the deployment ledger +- Confirm the RPC endpoint is correct for the network + +### Backfill hangs + +- Check RPC endpoint health: `curl -X POST $SOROBAN_RPC_URL -d '{"jsonrpc":"2.0","id":1,"method":"getHealth"}'` +- Verify network connectivity +- Check database connection pool: `SELECT * FROM pg_stat_activity;` + +### High memory usage + +- Reduce `--batch-size` (default 100) +- Increase checkpoint frequency (every 50 batches instead of 100) + +## Contract IDs (Reference) + +| Contract | Mainnet ID | Testnet ID | +| ---------------- | ---------- | ---------- | +| ArtistFacet | `CAAA...` | `CBBB...` | +| SongFacet | `CCCC...` | `CDDD...` | +| AlbumFacet | `CEEE...` | `CFFF...` | +| MarketplaceFacet | `CGGG...` | `CHHH...` | +| RoyaltyFacet | `CIII...` | `CJJJ...` | + +_(Replace with actual deployed contract IDs before use)_ + +## Next Steps + +After backfill completion: + +1. Start the indexer worker: `npm run worker:indexer` +2. Verify real-time event processing: `GET /api/admin/indexer/status` +3. Monitor metrics: http://localhost:3001/dashboards (Grafana) +4. Set up alerts for lag > 1000 ledgers or error rate > 0.1/s + +## Support + +For issues or questions: + +- Check logs: `docker logs audioblock_backend --tail 100` +- Query status: `GET /api/admin/indexer/status` +- Review metrics: http://localhost:3001 +- Open an issue: https://github.com/AudioBitsStellar/AudioBlock_Backend/issues diff --git a/monitoring/README.md b/monitoring/README.md index 0e75a36..7e14b18 100644 --- a/monitoring/README.md +++ b/monitoring/README.md @@ -58,6 +58,21 @@ Host/system overview using standard `prometheus-node-exporter` metrics node exporter is added to the compose topology and scraped — confirm the scrape job exists before relying on them. +### `audioblock-indexer.json` (Issue #242) + +Blockchain indexer health dashboard visualizing metrics from `IndexerService`: + +- `indexer_lag_ledgers` — Ledgers behind latest Stellar ledger (by network/contract) +- `indexer_events_processed_total` — Total events processed (cumulative and rate) +- `indexer_errors_total` — Error count and rate (with alerting) +- Event processing throughput (events/s) +- Lag distribution (p50/p95/p99) +- Per-network processing rate +- 24-hour summaries (events, errors) + +Includes alert rules for high lag (>1000 ledgers) and error rate (>0.1/s). +Covers all 5 AudioBlock contracts (Artist, Song, Album, Marketplace, Royalty). + ## Adding a dashboard 1. Create (or copy) a `*.json` file under `monitoring/dashboards/`. diff --git a/monitoring/dashboards/audioblock-indexer.json b/monitoring/dashboards/audioblock-indexer.json new file mode 100644 index 0000000..0811ee7 --- /dev/null +++ b/monitoring/dashboards/audioblock-indexer.json @@ -0,0 +1,215 @@ +{ + "__inputs": [], + "__requires": [], + "title": "AudioBlock Backend — Indexer Health", + "uid": "audioblock-indexer", + "version": 1, + "timezone": "utc", + "editable": true, + "refresh": "30s", + "schemaVersion": 30, + "panels": [ + { + "title": "Indexer Lag (ledgers behind)", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "targets": [ + { + "expr": "indexer_lag_ledgers", + "legendFormat": "{{network}}/{{contract}}", + "refId": "A" + } + ], + "yaxes": [ + { "format": "short", "label": "Ledgers Behind" }, + { "format": "short" } + ], + "alert": { + "name": "High Indexer Lag", + "message": "Indexer is falling behind", + "conditions": [ + { + "evaluator": { "type": "gt", "params": [1000] }, + "operator": { "type": "and" }, + "query": { "params": ["A", "5m", "now"] }, + "reducer": { "type": "avg" }, + "type": "query" + } + ] + } + }, + { + "title": "Event Processing Throughput (events/s)", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "targets": [ + { + "expr": "rate(indexer_events_processed_total[5m])", +: [ + { "format": "short", "label": "Total Events" }, + { "format": "short" } + ] + }, + { + "title": "Error Rate (errors/s)", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "targets": [ + { + "expr": "rate(indexer_errors_total[5m])", + "legendFormat": "{{network}}/{{contract}}", + "refId": "A" + } + ], + "yaxes": [ + { "format": "ops", "label": "Errors/s" }, + { "format": "short" } + ], + "alert": { + "name": "Indexer Errors", + "message": "Indexer is experiencing errors", + "conditions": [ + { + "evaluator": { "type": "gt", "params": [0.1] }, + "operator": { "type": "and" }, + "query": { "params": ["A", "5m", "now"] }, + "reducer": { "type": "avg" }, + "type": "query" + } + ] + } + }, + { + "title": "Total Errors by Contract", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "targets": [ + { + "expr": "indexer_errors_total", + "legendFormat": "{{network}}/{{contract}}", + "refId": "A" + } + ], + "yaxes": [ + { "format": "short", "label": "Total Errors" }, + { "format": "short" } + ] + }, + { + "title": "Lag by Contract (current)", + "type": "singlestat", + "gridPos": { "h": 4, "w": 4, "x": 12, "y": 16 }, + "targets": [ + { + "expr": "max(indexer_lag_ledgers) by (contract)", + "refId": "A" + } + ], + "format": "short", + "thresholds": "500,1000", + "colors": ["#299c46", "#e5ac0e", "#bf1b00"] + }, + { + "title": "Events Processed (24h)", + "type": "singlestat", + "gridPos": { "h": 4, "w": 4, "x": 16, "y": 16 }, + "targets": [ + { + "expr": "sum(increase(indexer_events_processed_total[24h]))", + "refId": "A" + } + ], + "format": "short" + }, + { + "title": "Errors (24h)", + "type": "singlestat", + "gridPos": { "h": 4, "w": 4, "x": 20, "y": 16 }, + "targets": [ + { + "expr": "sum(increase(indexer_errors_total[24h]))", + "refId": "A" + } + ], + "format": "short", + "thresholds": "10,50", + "colors": ["#299c46", "#e5ac0e", "#bf1b00"] + }, + { + "title": "Processing Rate by Network", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 20 }, + "targets": [ + { + "expr": "sum(rate(indexer_events_processed_total[5m])) by (network)", + "legendFormat": "{{network}}", + "refId": "A" + } + ], + "yaxes": [ + { "format": "ops", "label": "Events/s" }, + { "format": "short" } + ] + }, + { + "title": "Lag Distribution (all contracts)", + "type": "graph", + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 20 }, + "targets": [ + { + "expr": "histogram_quantile(0.50, sum(rate(indexer_lag_ledgers[5m])) by (le))", + "legendFormat": "p50", + "refId": "A" + }, + { + "expr": "histogram_quantile(0.95, sum(rate(indexer_lag_ledgers[5m])) by (le))", + "legendFormat": "p95", + "refId": "B" + }, + { + "expr": "histogram_quantile(0.99, sum(rate(indexer_lag_ledgers[5m])) by (le))", + "legendFormat": "p99", + "refId": "C" + } + ], + "yaxes": [ + { "format": "short", "label": "Lag (ledgers)" }, + { "format": "short" } + ] + } + ], + "templating": { + "list": [ + { + "name": "network", + "type": "query", + "query": "label_values(indexer_lag_ledgers, network)", + "current": { + "text": "All", + "value": "$__all" + }, + "includeAll": true + }, + { + "name": "contract", + "type": "query", + "query": "label_values(indexer_lag_ledgers{network=\"$network\"}, contract)", + "current": { + "text": "All", + "value": "$__all" + }, + "includeAll": true + } + ] + }, + "annotations": { + "list": [ + { + "name": "Deployments", + "datasource": "Prometheus", + "enable": true, + "iconColor": "blue" + } + ] + } +} diff --git a/src/__tests__/IndexerController.test.ts b/src/__tests__/IndexerController.test.ts new file mode 100644 index 0000000..791218e --- /dev/null +++ b/src/__tests__/IndexerController.test.ts @@ -0,0 +1,89 @@ +import { AdminController } from '../controllers/AdminController'; +import { createMockRequest, createMockResponse } from '../utils/testHelpers'; + +// Mock IndexerService +const mockGetAllStatus = jest.fn(); +const mockGetAllBackfillStatus = jest.fn(); + +jest.mock('../services/IndexerService', () => ({ +errorCount: 0, + lastError: null, + lastErrorAt: null, + lagLedgers: 10, + updatedAt: new Date('2026-08-31T10:00:00Z'), + }, + { + contractId: 'CBBB...SONG', + network: 'mainnet', + lastProcessedLedger: 950, + eventsProcessed: 320, + errorCount: 2, + lastError: 'RPC timeout', + lastErrorAt: new Date('2026-08-31T09:45:00Z'), + lagLedgers: 60, + updatedAt: new Date('2026-08-31T09:50:00Z'), + }, + ]; + + const mockBackfills = [ + { + contractId: 'CAAA...ARTIST', + network: 'mainnet', + completed: true, + startLedger: 1, + endLedger: 500, + eventsImported: 250, + errorMessage: null, + createdAt: new Date('2026-08-01T00:00:00Z'), + updatedAt: new Date('2026-08-01T12:00:00Z'), + }, + ]; + + mockGetAllStatus.mockResolvedValue(mockIndexers); + mockGetAllBackfillStatus.mockResolvedValue(mockBackfills); + + const req = createMockRequest({ query: { currentLedger: '1010' } }); + const res = createMockResponse(); + + await AdminController.getIndexerStatus(req, res as any); + + expect(mockGetAllStatus).toHaveBeenCalledWith(1010); + expect(mockGetAllBackfillStatus).toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(200); + expect(res.json).toHaveBeenCalledWith({ + success: true, + indexers: mockIndexers, + backfills: mockBackfills, + }); + }); + + it('returns status without currentLedger parameter', async () => { + mockGetAllStatus.mockResolvedValue([]); + mockGetAllBackfillStatus.mockResolvedValue([]); + + const req = createMockRequest({ query: {} }); + const res = createMockResponse(); + + await AdminController.getIndexerStatus(req, res as any); + + expect(mockGetAllStatus).toHaveBeenCalledWith(undefined); + expect(res.status).toHaveBeenCalledWith(200); + }); + + it('handles errors gracefully', async () => { + mockGetAllStatus.mockRejectedValue(new Error('Database connection failed')); + + const req = createMockRequest({ query: {} }); + const res = createMockResponse(); + + await AdminController.getIndexerStatus(req, res as any); + + expect(res.status).toHaveBeenCalledWith(500); + expect(res.json).toHaveBeenCalledWith( + expect.objectContaining({ + success: false, + message: expect.stringContaining('Database connection failed'), + }), + ); + }); +}); diff --git a/src/config/db.ts b/src/config/db.ts index 8333af5..c9d7869 100644 --- a/src/config/db.ts +++ b/src/config/db.ts @@ -12,6 +12,8 @@ import { TakedownRequest } from '../entities/TakedownRequest'; import { ApiKey } from '../entities/ApiKey'; import { AiGenerationRecord } from '../entities/AiGenerationRecord'; import { TweetDraft } from '../entities/TweetDraft'; +import { IndexerCursor } from '../entities/IndexerCursor'; +import { BackfillStatus } from '../entities/BackfillStatus'; dotenv.config(); @@ -38,6 +40,8 @@ const AppDataSource = new DataSource({ ApiKey, AiGenerationRecord, TweetDraft, + IndexerCursor, + BackfillStatus, ], migrations: [__dirname + '/../migrations/*.{js,ts}'], migrationsTableName: 'migrations', diff --git a/src/controllers/AdminController.ts b/src/controllers/AdminController.ts index 504a6cc..1eea424 100644 --- a/src/controllers/AdminController.ts +++ b/src/controllers/AdminController.ts @@ -16,6 +16,7 @@ export class AdminController { private static userService = new UserService(); private static artistProfileService = new ArtistProfileService(); private static transactionLogService = new TransactionLogService(); + private static indexerService = new (require('../services/IndexerService').IndexerService)(); /** * POST /api/admin/users/:id/role — assign a role to a user. @@ -156,4 +157,28 @@ export class AdminController { handleError(req, res, error); } }; + + /** + * GET /api/admin/indexer/status — get indexer health per contract/network (Issue #253). + * + * Returns cursor position, lag, event count, and last error for all contracts. + * Admin-gated via requirePermission middleware. + */ + static getIndexerStatus = async (req: Request, res: Response) => { + try { + const currentLedgerParam = req.query.currentLedger as string | undefined; + const currentLedger = currentLedgerParam ? Number(currentLedgerParam) : undefined; + + const statuses = await AdminController.indexerService.getAllStatus(currentLedger); + const backfillStatuses = await AdminController.indexerService.getAllBackfillStatus(); + + return res.status(HTTP_STATUS.OK).json({ + success: true, + indexers: statuses, + backfills: backfillStatuses, + }); + } catch (error) { + handleError(req, res, error); + } + }; } diff --git a/src/entities/BackfillStatus.ts b/src/entities/BackfillStatus.ts new file mode 100644 index 0000000..dd51f74 --- /dev/null +++ b/src/entities/BackfillStatus.ts @@ -0,0 +1,44 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, +} from 'typeorm'; + +/** + * Completion marker for one-time backfill operations (Issue #250). + * Prevents accidental re-execution of historical data imports. + */ +@Entity('backfill_status') +export class BackfillStatus { + @PrimaryGeneratedColumn('uuid') + id!: string; + + @Column({ type: 'varchar', length: 100 }) + contractId!: string; + + @Column({ type: 'varchar', length: 50 }) + network!: string; + + @Column({ type: 'boolean', default: false }) + completed!: boolean; + + @Column({ type: 'bigint', nullable: true }) + startLedger!: number | null; + + @Column({ type: 'bigint', nullable: true }) + endLedger!: number | null; + + @Column({ type: 'bigint', default: 0 }) + eventsImported!: number; + + @Column({ type: 'text', nullable: true }) + errorMessage!: string | null; + + @CreateDateColumn() + createdAt!: Date; + + @UpdateDateColumn() + updatedAt!: Date; +} diff --git a/src/entities/IndexerCursor.ts b/src/entities/IndexerCursor.ts new file mode 100644 index 0000000..60fc2ac --- /dev/null +++ b/src/entities/IndexerCursor.ts @@ -0,0 +1,25 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, +} from 'typeorm'; + +/** + * Tracks the indexer's last-processed ledger position per contract + network. + * Enables resumable event polling and historical backfill (Issues #241, #250, #253). + */ +@Entity('indexer_cursors') +export class IndexerCursor { + @PrimaryGeneratedColumn('uuid') + id!: string; + + @Column({ type: 'varchar', length: 100 }) + contractId!: string; + + @Column({ type: 'varchar', length: 50 }) + network!: string; // 'mainnet' | 'testnet' | 'futurenet' + + @Column({ type: 'bigint', default: 0 }) + lastProcessedLedger!: numbe diff --git a/src/migrations/1756684800000-AddIndexerEntities.ts b/src/migrations/1756684800000-AddIndexerEntities.ts new file mode 100644 index 0000000..46aaa02 --- /dev/null +++ b/src/migrations/1756684800000-AddIndexerEntities.ts @@ -0,0 +1,146 @@ +import { MigrationInterface, QueryRunner, Table, TableIndex } from 'typeorm'; + +/** + * Adds indexer_cursors and backfill_status tables for blockchain event indexing. + * Issues #241, #250, #253. + */ +export class AddIndexerEntities1756684800000 implements MigrationInterface { + public async up(queryR + length: '50', + }, + { + name: 'lastProcessedLedger', + type: 'bigint', + default: 0, + }, + { + name: 'eventsProcessed', + type: 'bigint', + default: 0, + }, + { + name: 'errorCount', + type: 'bigint', + default: 0, + }, + { + name: 'lastError', + type: 'text', + isNullable: true, + }, + { + name: 'lastErrorAt', + type: 'timestamp', + isNullable: true, + }, + { + name: 'createdAt', + type: 'timestamp', + default: 'CURRENT_TIMESTAMP', + }, + { + name: 'updatedAt', + type: 'timestamp', + default: 'CURRENT_TIMESTAMP', + }, + ], + }), + true, + ); + + // Add unique constraint on contractId + network + await queryRunner.createIndex( + 'indexer_cursors', + new TableIndex({ + name: 'IDX_indexer_cursors_contract_network', + columnNames: ['contractId', 'network'], + isUnique: true, + }), + ); + + // Create backfill_status table + await queryRunner.createTable( + new Table({ + name: 'backfill_status', + columns: [ + { + name: 'id', + type: 'uuid', + isPrimary: true, + default: 'uuid_generate_v4()', + }, + { + name: 'contractId', + type: 'varchar', + length: '100', + }, + { + name: 'network', + type: 'varchar', + length: '50', + }, + { + name: 'completed', + type: 'boolean', + default: false, + }, + { + name: 'startLedger', + type: 'bigint', + isNullable: true, + }, + { + name: 'endLedger', + type: 'bigint', + isNullable: true, + }, + { + name: 'eventsImported', + type: 'bigint', + default: 0, + }, + { + name: 'errorMessage', + type: 'text', + isNullable: true, + }, + { + name: 'createdAt', + type: 'timestamp', + default: 'CURRENT_TIMESTAMP', + }, + { + name: 'updatedAt', + type: 'timestamp', + default: 'CURRENT_TIMESTAMP', + }, + ], + }), + true, + ); + + // Add unique constraint on contractId + network + await queryRunner.createIndex( + 'backfill_status', + new TableIndex({ + name: 'IDX_backfill_status_contract_network', + columnNames: ['contractId', 'network'], + isUnique: true, + }), + ); + + // Add index on completed for quick filtering + await queryRunner.createIndex( + 'backfill_status', + new TableIndex({ + name: 'IDX_backfill_status_completed', + columnNames: ['completed'], + }), + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.dropTable('backfill_status'); + await queryRunner.dropTable('indexer_cursors'); + } +} diff --git a/src/routes/adminRoutes.ts b/src/routes/adminRoutes.ts index 43412da..d52b8a5 100644 --- a/src/routes/adminRoutes.ts +++ b/src/routes/adminRoutes.ts @@ -103,4 +103,11 @@ router.get( AdminController.getTransactionLogs, ); +// Indexer health/status (Issue #253) — admins and moderators. +router.get( + '/indexer/status', + requirePermission(Permission.CONTENT_MODERATE), + AdminController.getIndexerStatus, +); + export default router; diff --git a/src/services/IndexerService.ts b/src/services/IndexerService.ts new file mode 100644 index 0000000..5fd8d23 --- /dev/null +++ b/src/services/IndexerService.ts @@ -0,0 +1,245 @@ +/** + * Indexer service for tracking blockchain event processing state and metrics. + * Manages cursor positions, lag calculation, and health reporting (Issues #241, #253). + */ +import AppDataSource from '../config/db'; +import { IndexerCursor } from '../entities/IndexerCursor'; +import { BackfillStatus } from '../entities/BackfillStatus'; +import { indexerLagLedgers, indexerEventsProcessedTotal, indexerErrorsTotal } from './MetricsService'; +import logger from '../config/logger'; + +export interface IndexerStatus { + contractId: string; + network: string; + lastProcessedLedger: number; + eventsProcessed: number; + errorCount: number; + lastError: string | null; + lastErrorAt: Date | null; + lagLedgers: number; + updatedAt: Date; +} + +export interface BackfillInfo { + contractId: string; + network: string; + completed: boolean; + startLedger: number | null; + endLedger: number | null; + eventsImported: number; + errorMessage: string | null; + createdAt: Date; + updatedAt: Date; +} + +export class IndexerService { + private cursorRepo = AppDataSource.getRepository(IndexerCursor); + private backfillRepo = AppDataSource.getRepository(BackfillStatus); + + /** + * Get or create an indexer cursor for a contract + network pair. + */ + async getCursor(contractId: string, network: string): Promise { + let cursor = await this.cursorRepo.findOne({ + where: { contrac + eventCount: number = 1, + ): Promise { + const cursor = await this.getCursor(contractId, network); + cursor.lastProcessedLedger = ledger; + cursor.eventsProcessed += eventCount; + await this.cursorRepo.save(cursor); + + // Update Prometheus metrics + indexerEventsProcessedTotal.inc({ network, contract: contractId }, eventCount); + } + + /** + * Record an indexer error and update metrics. + */ + async recordError( + contractId: string, + network: string, + error: Error, + ): Promise { + const cursor = await this.getCursor(contractId, network); + cursor.errorCount += 1; + cursor.lastError = error.message; + cursor.lastErrorAt = new Date(); + await this.cursorRepo.save(cursor); + + // Update Prometheus metrics + indexerErrorsTotal.inc({ network, contract: contractId }); + } + + /** + * Calculate and update lag metrics for all cursors. + * Should be called periodically by a monitoring loop. + */ + async updateLagMetrics(currentLedger: number): Promise { + const cursors = await this.cursorRepo.find(); + + for (const cursor of cursors) { + const lag = Math.max(0, currentLedger - cursor.lastProcessedLedger); + indexerLagLedgers.set( + { network: cursor.network, contract: cursor.contractId }, + lag, + ); + } + } + + /** + * Get status for all indexers (admin endpoint, Issue #253). + */ + async getAllStatus(currentLedger?: number): Promise { + const cursors = await this.cursorRepo.find(); + + return cursors.map(cursor => ({ + contractId: cursor.contractId, + network: cursor.network, + lastProcessedLedger: cursor.lastProcessedLedger, + eventsProcessed: cursor.eventsProcessed, + errorCount: cursor.errorCount, + lastError: cursor.lastError, + lastErrorAt: cursor.lastErrorAt, + lagLedgers: currentLedger + ? Math.max(0, currentLedger - cursor.lastProcessedLedger) + : 0, + updatedAt: cursor.updatedAt, + })); + } + + /** + * Get status for a specific contract + network. + */ + async getStatus(contractId: string, network: string, currentLedger?: number): Promise { + const cursor = await this.getCursor(contractId, network); + + return { + contractId: cursor.contractId, + network: cursor.network, + lastProcessedLedger: cursor.lastProcessedLedger, + eventsProcessed: cursor.eventsProcessed, + errorCount: cursor.errorCount, + lastError: cursor.lastError, + lastErrorAt: cursor.lastErrorAt, + lagLedgers: currentLedger + ? Math.max(0, currentLedger - cursor.lastProcessedLedger) + : 0, + updatedAt: cursor.updatedAt, + }; + } + + // ── Backfill management (Issue #250) ────────────────────────────────────── + + /** + * Check if backfill has been completed for a contract + network. + */ + async isBackfillCompleted(contractId: string, network: string): Promise { + const status = await this.backfillRepo.findOne({ + where: { contractId, network }, + }); + return status?.completed ?? false; + } + + /** + * Mark backfill as started. + */ + async startBackfill( + contractId: string, + network: string, + startLedger: number, + endLedger: number, + ): Promise { + const existing = await this.backfillRepo.findOne({ + where: { contractId, network }, + }); + + if (existing?.completed) { + throw new Error( + `Backfill already completed for ${contractId} on ${network}. ` + + 'Delete the record manually if re-run is intentional.', + ); + } + + const status = existing || this.backfillRepo.create({ contractId, network }); + status.startLedger = startLedger; + status.endLedger = endLedger; + status.completed = false; + status.eventsImported = 0; + status.errorMessage = null; + + await this.backfillRepo.save(status); + logger.info({ contractId, network, startLedger, endLedger }, 'Backfill started'); + return status; + } + + /** + * Update backfill progress. + */ + async updateBackfillProgress( + contractId: string, + network: string, + eventsImported: number, + ): Promise { + const status = await this.backfillRepo.findOne({ + where: { contractId, network }, + }); + + if (status) { + status.eventsImported = eventsImported; + await this.backfillRepo.save(status); + } + } + + /** + * Mark backfill as completed. + */ + async completeBackfill(contractId: string, network: string): Promise { + const status = await this.backfillRepo.findOne({ + where: { contractId, network }, + }); + + if (status) { + status.completed = true; + await this.backfillRepo.save(status); + logger.info({ contractId, network }, 'Backfill completed'); + } + } + + /** + * Record backfill failure. + */ + async failBackfill( + contractId: string, + network: string, + error: Error, + ): Promise { + const status = await this.backfillRepo.findOne({ + where: { contractId, network }, + }); + + if (status) { + status.errorMessage = error.message; + await this.backfillRepo.save(status); + logger.error({ contractId, network, error }, 'Backfill failed'); + } + } + + /** + * Get all backfill statuses (for admin visibility). + */ + async getAllBackfillStatus(): Promise { + const statuses = await this.backfillRepo.find(); + return statuses.map(s => ({ + contractId: s.contractId, + network: s.network, + completed: s.completed, + startLedger: s.startLedger, + endLedger: s.endLedger, + eventsImported: s.eventsImported, + errorMessage: s.errorMessage, + createdAt: s.createdAt, + updatedAt: s.updatedAt, + })); + } +} diff --git a/src/services/MetricsService.ts b/src/services/MetricsService.ts index 66bda53..426ac62 100644 --- a/src/services/MetricsService.ts +++ b/src/services/MetricsService.ts @@ -1,6 +1,13 @@ /** * Prometheus metrics for monitoring HTTP requests, database pool, uploads, - * royalties, marketplace volume, and cache performance. + * royalties, marketplace volume, cache performance, and blockchain indexer health. + * + * Indexer metrics (Issues #241, #242): + * - indexer_lag_ledgers: Ledgers behind latest Stellar ledger (by network/contract) + * - indexer_events_processed_total: Total blockchain events processed + * - indexer_errors_total: Total indexer errors + * + * See monitoring/dashboards/audioblock-indexer.json for visualization. */ import client from 'prom-client'; @@ -77,6 +84,29 @@ export const cacheMissesTotal = new client.Counter({ registers: [register], }); +// ── Indexer metrics (Issue #241) ──────────────────────────────────────────── + +export const indexerLagLedgers = new client.Gauge({ + name: 'indexer_lag_ledgers', + help: 'Number of ledgers behind the latest Stellar ledger', + labelNames: ['network', 'contract'] as const, + registers: [register], +}); + +export const indexerEventsProcessedTotal = new client.Counter({ + name: 'indexer_events_processed_total', + help: 'Total number of blockchain events processed by the indexer', + labelNames: ['network', 'contract'] as const, + registers: [register], +}); + +export const indexerErrorsTotal = new client.Counter({ + name: 'indexer_errors_total', + help: 'Total number of indexer errors', + labelNames: ['network', 'contract'] as const, + registers: [register], +}); + /** * Update the Prometheus gauges reflecting the current PostgreSQL connection * pool state.