Skip to content

Commit f9c5d99

Browse files
ayomideadeniranayomideadeniran
authored
Merge pull request #1130 from danieloche635-bit/fix/indexer-metrics-migrate-integrity-796-799-800-801
fix: indexer observability, integrity audit & Postgres/TimescaleDB migrations (#796 #799 #800 #801)
2 parents 3748737 + 9ea7dc6 commit f9c5d99

11 files changed

Lines changed: 735 additions & 0 deletions

‎indexer/docs/grafana-indexer-dashboard.json‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
{
2+
"__inputs": [],
3+
"annotations": {
4+
"list": []
5+
},
6+
"editable": true,
7+
"fiscalYearStartMonth": 0,
8+
"graphTooltip": 1,
9+
"id": null,
10+
"links": [],
11+
"panels": [
12+
{
13+
"datasource": { "uid": "${DS_PROMETHEUS}" },
14+
"fieldConfig": {
15+
"defaults": {
16+
"color": { "mode": "thresholds" },
17+
"custom": {
18+
"drawStyle": "line",
19+
"fillOpacity": 10,
20+
"lineWidth": 2,
21+
"pointSize": 5,
22+
"showPoints": "never",
23+
"spanNulls": true
24+
},
25+
"thresholds": {
26+
"mode": "absolute",
27+
"steps": [
28+
{ "color": "green", "value": null },
29+
{ "color": "orange", "value": 25 },
30+
{ "color": "red", "value": 50 }
31+
]
32+
},
33+
"unit": "ledgers"
34+
},
35+
"overrides": []
36+
},
37+
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
38+
"id": 1,
39+
"options": {
40+
"legend": { "calcs": ["last"], "displayMode": "list", "placement": "bottom" },
41+
"tooltip": { "mode": "multi", "sort": "desc" }
42+
},
43+
"targets": [{ "expr": "indexer_lag_ledgers", "legendFormat": "lag" }],
44+
"title": "Indexer Lag (ledgers behind network head)",
45+
"type": "timeseries"
46+
},
47+
{
48+
"datasource": { "uid": "${DS_PROMETHEUS}" },
49+
"fieldConfig": {
50+
"defaults": {
51+
"color": { "mode": "palette-classic" },
52+
"custom": {
53+
"drawStyle": "line",
54+
"fillOpacity": 0,
55+
"lineWidth": 2,
56+
"showPoints": "never",
57+
"spanNulls": true
58+
},
59+
"unit": "short"
60+
},
61+
"overrides": []
62+
},
63+
"gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
64+
"id": 2,
65+
"options": {
66+
"legend": { "calcs": ["last"], "displayMode": "list", "placement": "bottom" },
67+
"tooltip": { "mode": "multi", "sort": "none" }
68+
},
69+
"targets": [
70+
{ "expr": "indexer_ledger_head", "legendFormat": "indexer head" },
71+
{ "expr": "network_ledger_head", "legendFormat": "network head" }
72+
],
73+
"title": "Ledger Heads",
74+
"type": "timeseries"
75+
},
76+
{
77+
"datasource": { "uid": "${DS_PROMETHEUS}" },
78+
"fieldConfig": {
79+
"defaults": {
80+
"color": { "mode": "palette-classic" },
81+
"custom": {
82+
"drawStyle": "bars",
83+
"fillOpacity": 40,
84+
"lineWidth": 1,
85+
"showPoints": "never"
86+
},
87+
"unit": "events/s"
88+
},
89+
"overrides": []
90+
},
91+
"gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 },
92+
"id": 3,
93+
"options": {
94+
"legend": { "calcs": ["lastNotNone"], "displayMode": "table", "placement": "bottom" },
95+
"tooltip": { "mode": "multi", "sort": "desc" }
96+
},
97+
"targets": [
98+
{
99+
"expr": "sum by (event_name) (rate(events_indexed_total[5m]))",
100+
"legendFormat": "{{event_name}}"
101+
}
102+
],
103+
"title": "Ingestion Rate by Event Type (events/s)",
104+
"type": "timeseries"
105+
},
106+
{
107+
"datasource": { "uid": "${DS_PROMETHEUS}" },
108+
"fieldConfig": {
109+
"defaults": {
110+
"color": { "mode": "thresholds" },
111+
"custom": { "fillOpacity": 80 },
112+
"min": 0,
113+
"thresholds": {
114+
"mode": "absolute",
115+
"steps": [
116+
{ "color": "green", "value": null },
117+
{ "color": "red", "value": 1 }
118+
]
119+
},
120+
"unit": "ledgers"
121+
},
122+
"overrides": []
123+
},
124+
"gridPos": { "h": 8, "w": 7, "x": 12, "y": 8 },
125+
"id": 4,
126+
"options": {
127+
"orientation": "horizontal",
128+
"reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
129+
"showThresholdLabels": false,
130+
"showThresholdMarkers": true
131+
},
132+
"targets": [{ "expr": "indexer_lag_ledgers", "legendFormat": "current lag" }],
133+
"title": "Current Lag",
134+
"type": "gauge"
135+
}
136+
],
137+
"refresh": "30s",
138+
"schemaVersion": 39,
139+
"tags": ["indexer", "sorotask", "stellar"],
140+
"templating": {
141+
"list": []
142+
},
143+
"time": { "from": "now-6h", "to": "now" },
144+
"timepicker": {},
145+
"timezone": "browser",
146+
"title": "SoroTask Indexer",
147+
"uid": "sorotask-indexer",
148+
"version": 1,
149+
"weekStart": ""
150+
}

‎indexer/docs/metrics.md‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
# Indexer Prometheus Metrics
2+
3+
The SoroTask indexer exposes a Prometheus `/metrics` endpoint (via
4+
[`prom-client`](../src/metrics.js)) for operators to scrape. A ready-to-import
5+
Grafana dashboard lives at [`grafana-indexer-dashboard.json`](./grafana-indexer-dashboard.json).
6+
7+
## Scrape configuration
8+
9+
```yaml
10+
scrape_configs:
11+
- job_name: sorotask-indexer
12+
static_configs:
13+
- targets: ["indexer:4000"]
14+
metrics_path: /metrics
15+
```
16+
17+
## Exported metrics
18+
19+
| Metric | Type | Help |
20+
| --- | --- | --- |
21+
| `indexer_ledger_head` | Gauge | Current ledger sequence processed by the indexer |
22+
| `network_ledger_head` | Gauge | Latest ledger sequence on the Stellar network head |
23+
| `indexer_lag_ledgers` | Gauge | Number of ledgers the indexer is lagging behind the network head |
24+
| `events_indexed_total` | Counter | Total contract events indexed, labeled by `event_name` |
25+
| `indexer_*` (defaults) | — | Node/process metrics (event loop, memory, GC) via `collectDefaultMetrics` |
26+
27+
### Deriving ingestion rate
28+
29+
`events_indexed_total` is a monotonically increasing counter, so ingestion
30+
throughput is computed with a `rate()` query:
31+
32+
```promql
33+
sum by (event_name) (rate(events_indexed_total[5m]))
34+
```
35+
36+
### Alerting on lag
37+
38+
A common alert triggers when the indexer falls behind the network head:
39+
40+
```promql
41+
indexer_lag_ledgers > 50
42+
```
43+
44+
See [`../../../deploy/grafana/`](https://github.com/SoroLabs/SoroTask) for
45+
deployment-level alerting conventions.

‎indexer/migrations/002_timescaledb_raw_events_retention.down.sql‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
-- Down migration: 002_timescaledb_raw_events_retention
2+
--
3+
-- Reverses 002_timescaledb_raw_events_retention.sql back to the plain `events`
4+
-- table that 001_initial_schema created. Because the up migration renamed the
5+
-- table and exposed it through a compatibility view, the down path reverses
6+
-- those in dependency order: drop the view, detach from TimescaleDB by removing
7+
-- the hypertable's ancillary objects, then rename raw_events back to events.
8+
--
9+
-- Guarded with IF EXISTS / defensive checks so a partial earlier state can
10+
-- still be rolled back.
11+
12+
SELECT remove_compression_policy('raw_events', if_exists => TRUE);
13+
14+
DROP VIEW IF EXISTS events;
15+
16+
DROP INDEX IF EXISTS idx_raw_events_ledger_timestamp;
17+
18+
ALTER TABLE raw_events SET (timescaledb.compress = 'false');
19+
20+
ALTER TABLE raw_events DROP CONSTRAINT IF EXISTS raw_events_dedup_key;
21+
ALTER TABLE raw_events DROP COLUMN IF EXISTS ledger_timestamp;
22+
23+
-- Restore the original table name used by 001_initial_schema.
24+
ALTER TABLE IF EXISTS raw_events RENAME TO events;

‎indexer/migrations/004_contract_versioning.down.sql‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
-- Down migration: 004_contract_versioning
2+
3+
DROP INDEX IF EXISTS idx_raw_events_wasm_version;
4+
5+
ALTER TABLE raw_events DROP COLUMN IF EXISTS contract_version;
6+
ALTER TABLE raw_events DROP COLUMN IF EXISTS wasm_version_hash;
7+
8+
ALTER TABLE tasks DROP COLUMN IF EXISTS contract_version;
9+
ALTER TABLE tasks DROP COLUMN IF EXISTS wasm_version_hash;
10+
11+
DROP TABLE IF EXISTS contract_versions;

‎indexer/migrations/004_contract_versioning.sql‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
-- Migration: 004_contract_versioning
2+
--
3+
-- Version indexer schema definitions against the on-chain contract WASM
4+
-- version hash (issue #799). A smart contract ABI upgrade is only safe once
5+
-- the indexer can attribute each event to the exact WASM artifact that emitted
6+
-- it — and can route that event to the matching schema version (V1 or V2).
7+
--
8+
-- Backward compatible by construction: the new columns are nullable with
9+
-- defaults, so existing V1 event rows and current insert paths keep working
10+
-- while the new fields are populated incrementally.
11+
12+
CREATE TABLE IF NOT EXISTS contract_versions (
13+
id BIGSERIAL PRIMARY KEY,
14+
contract_id TEXT NOT NULL,
15+
wasm_version_hash TEXT NOT NULL,
16+
schema_version TEXT NOT NULL DEFAULT 'v1',
17+
start_ledger BIGINT,
18+
end_ledger BIGINT,
19+
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
20+
UNIQUE (contract_id, wasm_version_hash)
21+
);
22+
23+
-- Attribute each stored event (and each indexed task) to the emitting WASM
24+
-- artifact. Existing rows default to 'v1' and a null hash until re-hydrated.
25+
ALTER TABLE raw_events ADD COLUMN IF NOT EXISTS wasm_version_hash TEXT;
26+
ALTER TABLE raw_events ADD COLUMN IF NOT EXISTS contract_version TEXT DEFAULT 'v1';
27+
28+
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS wasm_version_hash TEXT;
29+
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS contract_version TEXT DEFAULT 'v1';
30+
31+
CREATE INDEX IF NOT EXISTS idx_raw_events_wasm_version
32+
ON raw_events (wasm_version_hash, ledger_sequence DESC);
33+
34+
INSERT INTO schema_migrations (version)
35+
VALUES ('004_contract_versioning')
36+
ON CONFLICT (version) DO NOTHING;

‎indexer/migrations/005_timescaledb_execution_continuous_aggregates.down.sql‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
-- Down migration: 005_timescaledb_execution_continuous_aggregates
2+
--
3+
-- Removes the continuous aggregates and their refresh policies, then drops the
4+
-- hypertable-specific index. The `executions` table itself is retained (its
5+
-- creation belongs to 001_initial_schema), but reverting 005 makes it a plain
6+
-- table again.
7+
8+
SELECT remove_continuous_aggregate_policy('execution_hourly', if_exists => TRUE);
9+
SELECT remove_continuous_aggregate_policy('execution_daily', if_exists => TRUE);
10+
11+
DROP MATERIALIZED VIEW IF EXISTS execution_hourly;
12+
DROP MATERIALIZED VIEW IF EXISTS execution_daily;
13+
14+
DROP INDEX IF EXISTS idx_executions_executed_at_ts;

‎indexer/migrations/005_timescaledb_execution_continuous_aggregates.sql‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
-- Migration: 005_timescaledb_execution_continuous_aggregates
2+
--
3+
-- Builds the time-series analytics surface for execution metrics (issue #796).
4+
-- `executions` becomes a TimescaleDB hypertable over executed_at, and hourly +
5+
-- daily continuous aggregates pre-aggregate throughput and fee metrics so
6+
-- dashboard queries never scan the raw table. Requires TimescaleDB.
7+
--
8+
-- Applied after 001 (executions) and 002 (TimescaleDB enabled + raw_events
9+
-- hypertable), so only this migration guarantees the executions hypertable.
10+
11+
CREATE INDEX IF NOT EXISTS idx_executions_executed_at_ts
12+
ON executions (executed_at DESC);
13+
14+
SELECT create_hypertable(
15+
'executions',
16+
'executed_at',
17+
chunk_time_interval => INTERVAL '7 days',
18+
migrate_data => TRUE,
19+
if_not_exists => TRUE
20+
);
21+
22+
-- Hourly throughput + fee analytics per task and per status.
23+
CREATE MATERIALIZED VIEW IF NOT EXISTS execution_hourly
24+
WITH (timescaledb.continuous) AS
25+
SELECT time_bucket('1 hour', executed_at) AS bucket,
26+
task_id,
27+
status,
28+
COUNT(*) AS executions,
29+
SUM(fee_paid) AS fee_paid_total
30+
FROM executions
31+
GROUP BY bucket, task_id, status
32+
WITH NO DATA;
33+
34+
SELECT add_continuous_aggregate_policy(
35+
'execution_hourly',
36+
start_offset => INTERVAL '3 days',
37+
end_offset => INTERVAL '1 hour',
38+
schedule_interval => INTERVAL '1 hour',
39+
if_not_exists => TRUE
40+
);
41+
42+
-- Daily throughput + average fee analytics per task.
43+
CREATE MATERIALIZED VIEW IF NOT EXISTS execution_daily
44+
WITH (timescaledb.continuous) AS
45+
SELECT time_bucket('1 day', executed_at) AS bucket,
46+
task_id,
47+
COUNT(*) AS executions,
48+
SUM(fee_paid) AS fee_paid_total,
49+
AVG(fee_paid) AS avg_fee_paid
50+
FROM executions
51+
GROUP BY bucket, task_id
52+
WITH NO DATA;
53+
54+
SELECT add_continuous_aggregate_policy(
55+
'execution_daily',
56+
start_offset => INTERVAL '30 days',
57+
end_offset => INTERVAL '1 hour',
58+
schedule_interval => INTERVAL '1 day',
59+
if_not_exists => TRUE
60+
);
61+
62+
INSERT INTO schema_migrations (version)
63+
VALUES ('005_timescaledb_execution_continuous_aggregates')
64+
ON CONFLICT (version) DO NOTHING;

‎indexer/src/index.js‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ const { runStaleTaskCleanup } = require("./staleTasks");
99
const { startApiServer } = require("./api");
1010
const { broadcastEvent } = require("./wsServer");
1111
const { computeAndStoreLedgerMerkle } = require("./merkleStore");
12+
const { LedgerAuditor, ensureAuditSchema } = require("./ledgerAuditor");
1213
const { scheduleArchival } = require("./archival");
1314
const { pubsub, EVENT_ADDED } = require("./graphql/pubsub");
1415
const { LedgerHashValidator } = require("./ledgerHashValidator");
@@ -683,6 +684,20 @@ if (!handleCLI()) {
683684
console.log("Starting periodic reconciliation (every 5 minutes)...");
684685
setInterval(reconcileAll, RECONCILE_INTERVAL_MS);
685686

687+
// Issue #800: background ledger integrity audit — recompute per-ledger Merkle
688+
// roots and alert operators the moment the store diverges from what was
689+
// anchored at ingest time (catches silent corruption / parser bugs).
690+
ensureAuditSchema(dbDeps).catch((err) => {
691+
console.error("[LedgerAuditor] Schema init error:", err.message);
692+
});
693+
const ledgerAuditor = new LedgerAuditor({
694+
deps: dbDeps,
695+
rpc,
696+
intervalMs: Number(process.env.AUDIT_INTERVAL_MS || 15 * 60 * 1000),
697+
maxLedgers: Number(process.env.AUDIT_MAX_LEDGERS || 64),
698+
});
699+
ledgerAuditor.start();
700+
686701
// Start synthetic transaction monitoring for end-to-end ingestion health
687702
const syntheticMonitor = new SyntheticMonitor({
688703
db,

0 commit comments

Comments
 (0)