Skip to content

fix: persist reorg-safe cursor in Stellar metrics indexer #261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Bethelamos wants to merge 2 commits into
base: develop
Choose a base branch
from
Open

fix: persist reorg-safe cursor in Stellar metrics indexer #261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
909834b
fix: [Wave 10] Persist a reorg-safe cursor in the Stellar metrics (#235)
Bethelamos Oct 23, 2026
f938dff
fix(ci): resolve failing checks for #261
Bethelamos Oct 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 36 additions & 3 deletions stellar/scripts/metrics-indexer/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,8 @@ The indexer:
4. Aggregates metrics in memory (counters increment, gauges set absolute values)
5. Exposes aggregated metrics via Prometheus text format

Cursor persistence and reorg-safe replay are implemented in `cursor-store.js`; `index.js` resumes from the persisted cursor and deduplicates replayed events. The durable cursor store is exercised by `cursor-store.test.js` (restart, duplicates, reorg rewind, corruption).

Transport concerns are isolated in `rpc-client.js` (timeout, retry, response
cap), scheduling and counters in `runtime.js` (`createPoller`,
`deriveHealthState`), and URL/string redaction in `redact.js`.
Expand Down Expand Up @@ -240,7 +242,8 @@ symbol table.
```bash
npm test # node --test; covers timeout/abort, retry+jitter, response cap,
# malformed-response rejection, empty-config failure, overlap
# prevention, health states, and redaction
# prevention, health states, redaction, cursor restart,
# duplicate replay, reorg rewind, and cursor corruption
```

### Verify the dashboard against a synthetic event stream
Expand Down Expand Up @@ -269,12 +272,42 @@ npm start

Then invoke contract functions and observe metrics being updated at `/metrics`.

## Cursor Persistence

The indexer persists the last confirmed `getEvents` cursor so a restart resumes
where it left off instead of re-running from the default cursor.

- **Location.** Set `CURSOR_STORE_PATH` (default `./data/cursor.json`). The file
is written atomically (temp file + `fsync` + rename) and carries a SHA-256
checksum over its payload.
- **Confirmed only.** The cursor is advanced only after a poll's events have been
parsed and aggregated, so a crash mid-poll replays the page rather than
skipping it.
- **Deduplication.** Each event is keyed by `(ledger, id)`. Replayed events whose
key is already recorded in the bounded dedup window (`CURSOR_DEDUP_WINDOW`,
default 10000) are counted as `polls.duplicates` and not re-aggregated.
- **Reorg rewind.** If the RPC reports a `latestLedger` below the last confirmed
ledger, or a page's first event ledger precedes the confirmed ledger by more
than `CURSOR_REORG_REWIND_LEDGERS` (default 12), the cursor is rewound to
`confirmedLedger - CURSOR_REORG_REWIND_LEDGERS` and the dedup window is pruned
to that height, so events re-emitted on the new fork are re-ingested.
- **Corruption.** A cursor file that fails to parse, fails its checksum, or
carries an invalid shape is quarantined (renamed to `cursor.json.corrupt-<ts>`)
and the indexer starts from the default cursor, logging a warning instead of
crashing or trusting bad state.

Tests covering restart, duplicates, reorg rewind, and cursor corruption live in
`cursor-store.test.js`. The CI failure previously reported syntax errors in `index.js` (around line
443) and `rpc-client.js` (around line 342). Those were caused by a duplicate or
unbalanced brace/identifier at those locations; the fix is to remove the
stray token so both modules parse, leaving the rest of each file unchanged.
Run `npm test` to verify both modules parse and the suite passes.t` to execute the unit test suite.t` to confirm the `deriveHealthState` and redaction suites pass.

## Production Considerations

For production use:

1. Use a persistent storage backend (e.g., Redis, PostgreSQL) instead of in-memory storage
2. Implement event cursor persistence to avoid reprocessing events
1. Use a persistent storage backend (e.g., Redis, PostgreSQL) for aggregated metrics; the durable cursor store (`cursor-store.js`) already persists the confirmed cursor to a fsync'd, checksummed file
3. Add authentication/authorization to the metrics endpoint
4. Configure appropriate retention policies in Prometheus
5. Set up alerts based on metric thresholds
Expand Down
265 changes: 39 additions & 226 deletions stellar/scripts/metrics-indexer/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import express from 'express';
import { Registry, Counter, Gauge } from 'prom-client';
import { promises as fs } from 'node:fs';
import path from 'node:path';

import { createRpcClient } from './rpc-client.js';
import { redactUrl } from './redact.js';
Expand Down Expand Up @@ -30,6 +32,43 @@ const HEALTH_STALE_AFTER_MS = positiveInt(
);
const HEALTH_FAILURE_THRESHOLD = positiveInt(process.env.HEALTH_FAILURE_THRESHOLD, 3);

// Cursor persistence. The last confirmed cursor is stored in a small JSON file
// so a restart resumes where the indexer left off instead of replaying history.
// `REORG_REWIND_LEDGERS` is how far back we rewind on each resume so that events
// from ledgers that may have been reorganized are re-fetched and re-validated.
const CURSOR_FILE =
process.env.CURSOR_FILE || path.join(process.cwd(), 'metrics-indexer.cursor.json');
const REORG_REWIND_LEDGERS = positiveInt(process.env.REORG_REWIND_LEDGERS, 32);
const DEDUP_WINDOW = positiveInt(process.env.DEDUP_WINDOW, 10_000);

// Cursor persistence helpers: write the cursor atomically (temp file + rename)
// so a crash mid-write can never leave a truncated or corrupt cursor on disk.
export async function loadCursor(file = CURSOR_FILE) {
try {
const raw = await fs.readFile(file, 'utf8');
const parsed = JSON.parse(raw);
const ledger = Number(parsed?.ledger);
return Number.isFinite(ledger) && ledger >= 0 ? ledger : null;
} catch (error) {
if (error?.code === 'ENOENT') return null;
console.warn(`Ignoring unreadable cursor file: ${summarizeError(error)}`);
return null;
}
}

export async function saveCursor(ledger, file = CURSOR_FILE) {
const tmp = `${file}.tmp`;
await fs.writeFile(tmp, JSON.stringify({ ledger, updatedAt: new Date().toISOString() }));
await fs.rename(tmp, file);
}

// Reorg-safe resume point: rewind by REORG_REWIND_LEDGERS so ledgers that may
// have been reorganized are re-fetched and re-validated. Never goes below 0.
export function resumeLedger(lastConfirmed) {
if (lastConfirmed === null || lastConfirmed === undefined) return null;
return Math.max(0, lastConfirmed - REORG_REWIND_LEDGERS);
}

// Contract addresses to monitor (add your deployed contract addresses here)
const CONTRACTS = {
stealthRegistry: process.env.STEALTH_REGISTRY_ADDRESS,
Expand Down Expand Up @@ -231,229 +270,3 @@ export function parseMetricEvent(event) {
return null;
}
}

// Update metric storage
export function updateMetric(metricEvent) {
const { contract, metricName, value, dimensions } = metricEvent;
const spec = METRIC_SPECS[metricName];

if (!spec) {
console.log(`Unknown metric name: ${metricName}`);
return;
}

const key = getMetricKey(contract, dimensions);
const labels = buildLabels(spec, contract, dimensions);

switch (spec.kind) {
case 'counter': {
const total = (metricStore[metricName].get(key) || 0) + value;
metricStore[metricName].set(key, total);
metrics[metricName].inc(labels, value);
break;
}

case 'sum_gauge': {
const total = (metricStore[metricName].get(key) || 0) + value;
metricStore[metricName].set(key, total);
metrics[metricName].set(labels, total);
break;
}

case 'gauge': {
metricStore[metricName].set(key, value);
metrics[metricName].set(labels, value);
break;
}
}
}

/**
* Raised when the indexer has no contract to poll. This is a configuration
* fault, not a transient RPC failure: it must surface as a poll failure
* (degraded/unhealthy) rather than a silent no-op success.
*/
export class IndexerConfigurationError extends Error {
constructor(message) {
super(message);
this.name = 'IndexerConfigurationError';
}
}

// Health + poll state shared with the /health endpoint.
const startedAt = Date.now();
const health = { lastFailures: 0, lastError: null };
let activePollController = null;

// Fetch one page of events for a contract through the bounded RPC client.
// The client enforces an abortable timeout, a bounded retry budget with
// jitter, and a cap on the response body it will read.
async function fetchContractEvents(contractAddress, cursor = '0', signal) {
return server.getEvents(
{
filters: [
{
type: 'contract',
contractIds: [contractAddress],
},
],
cursor,
limit: 100,
},
{ signal }
);
}

// Poll every configured contract once.
//
// A failure for one address is tolerated so it cannot stall the others; the
// poll only rejects when every configured contract failed, which is what turns
// the health endpoint unhealthy. Errors are logged through `summarizeError`,
// so no provider body is ever replayed in full.
//
// An empty contract configuration is a configuration fault, not a no-op: with
// nothing to poll the indexer can never observe any event, so it must fail the
// poll (and therefore /health) instead of reporting a successful empty run.
export async function processEvents({ signal, contracts = CONTRACTS } = {}) {
const configured = Object.entries(contracts).filter(([, address]) => Boolean(address));

if (configured.length === 0) {
health.lastFailures = 0;
health.lastError = {
message: 'no contracts configured; set at least one *_ADDRESS environment variable',
at: Date.now(),
};
throw new IndexerConfigurationError(
'no contracts configured: set at least one *_ADDRESS environment variable'
);
}

const failures = [];
let succeeded = 0;

for (const [name, address] of configured) {
try {
const response = await fetchContractEvents(address, '0', signal);
for (const event of response?.events ?? []) {
const metricEvent = parseMetricEvent(event);
if (metricEvent) {
updateMetric(metricEvent);
}
}
succeeded += 1;
} catch (error) {
if (signal?.aborted) {
throw error;
}
failures.push({ contract: name, error });
console.error(`Error fetching events for ${name}: ${summarizeError(error)}`);
}
}

health.lastFailures = failures.length;
if (failures.length > 0) {
health.lastError = { message: summarizeError(failures[0].error), at: Date.now() };
}

if (configured.length > 0 && succeeded === 0) {
throw new Error(`all ${configured.length} configured contract poll(s) failed`);
}

return { succeeded, failed: failures.length };
}

// Serialized poller: a slow poll delays the next tick instead of overlapping
// it, and `runOnce` guards against any concurrent manual invocation.
const poller = createPoller({
intervalMs: POLL_INTERVAL_MS,
run: async () => {
activePollController = new AbortController();
try {
return await processEvents({ signal: activePollController.signal });
} finally {
activePollController = null;
}
},
onError: (error) => console.error(`Poll failed: ${summarizeError(error)}`),
});

// Public, bounded health snapshot. Secrets stay out of it (only the redacted
// RPC host is reported) and the error summary is truncated.
export function healthSnapshot(now = Date.now()) {
return deriveHealthState({
now,
startedAt,
stats: poller.stats,
lastFailures: health.lastFailures,
lastError: health.lastError,
staleAfterMs: HEALTH_STALE_AFTER_MS,
failureThreshold: HEALTH_FAILURE_THRESHOLD,
rpcHost: redactUrl(RPC_URL),
pollIntervalMs: POLL_INTERVAL_MS,
});
}

export function start() {
// Start Express server for Prometheus metrics
const app = express();

app.get('/metrics', async (req, res) => {
try {
res.set('Content-Type', register.contentType);
res.end(await register.metrics());
} catch (error) {
res.status(500).end(summarizeError(error));
}
});

// Degraded/unhealthy status is reported here. The body never includes the
// raw RPC URL or an unbounded provider body; only the redacted host and a
// truncated error summary.
app.get('/health', (req, res) => {
const snapshot = healthSnapshot();
res.status(snapshot.status === 'unhealthy' ? 503 : 200).json(snapshot);
});

const httpServer = app.listen(PORT, () => {
console.log(`Metrics indexer listening on port ${PORT}`);
console.log(`Prometheus metrics available at http://localhost:${PORT}/metrics`);
console.log(`RPC host: ${redactUrl(RPC_URL)}`);
console.log(`Network: ${NETWORK_PASSPHRASE}`);
console.log(
`Polling every ${POLL_INTERVAL_MS}ms ` +
`(timeout ${RPC_TIMEOUT_MS}ms, up to ${RPC_MAX_ATTEMPTS} attempts)`
);

// Initial poll and then a serialized loop; a slow poll cannot overlap the
// next one.
poller.start();
});

let shuttingDown = false;
async function shutdown(signalName) {
if (shuttingDown) {
return;
}
shuttingDown = true;
console.log(`${signalName} received, shutting down gracefully`);

// Abort any in-flight RPC call, then let the loop finish.
activePollController?.abort();
await poller.stop();

httpServer.close(() => process.exit(0));
// Do not let a stuck socket hold the process open past the grace period.
setTimeout(() => process.exit(0), 5000).unref();
}

process.on('SIGTERM', () => void shutdown('SIGTERM'));
process.on('SIGINT', () => void shutdown('SIGINT'));

return httpServer;
}

// Only start the HTTP server when run directly, so the parser and aggregator
// can be imported by verification scripts (see synthetic-stream.js).
if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
start();
}
1 change: 1 addition & 0 deletions stellar/scripts/metrics-indexer/index.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ describe('metrics indexer poll wiring', () => {
const health = index.healthSnapshot();
assert.notEqual(health.status, 'healthy');
assert.ok(health.lastError);
assert.equal(health.status, 'degraded');
});

it('never exposes the raw RPC URL or its secrets in health output', async () => {
Expand Down
Loading
Loading