diff --git a/admin/remotetables/setup.go b/admin/remotetables/setup.go index ad3ecaf22..1f748ca91 100644 --- a/admin/remotetables/setup.go +++ b/admin/remotetables/setup.go @@ -36,6 +36,10 @@ var externalRemoteTables = []struct { {"dzdp", "offsets"}, {"dzdp", "location_decisions"}, {"dzdp", "location_state"}, + // lagged_validators is filled by an external process and only exists in prod. + // Proxy just this table so non-prod envs can read it; validator_rewards in the + // same database is ingested natively by the indexer and is left untouched. + {"dzf_data", "lagged_validators"}, } // externalRemoteDatabases lists remote databases to mirror in full, discovering diff --git a/api/config/config.go b/api/config/config.go index 5255ee83f..71907a440 100644 --- a/api/config/config.go +++ b/api/config/config.go @@ -35,6 +35,9 @@ var publisherDB = "shredder" // dzdpDB is the ClickHouse database name for DZDP tables e.g. location_state (default: "dzdp"). var dzdpDB = "dzdp" +// dzfDataDB is the ClickHouse database name for dzf_data tables (default: "dzf_data"). +var dzfDataDB = "dzf_data" + // EnvDBs maps environment names to their ClickHouse connection pools. // The mainnet-beta entry always points to DB. var EnvDBs map[string]driver.Conn @@ -83,6 +86,16 @@ func SetDZDPDB(db string) { dzdpDB = db } +// GetDZFDataDB returns the dzf_data database name. +func GetDZFDataDB() string { + return dzfDataDB +} + +// SetDZFDataDB sets the dzf_data database name. +func SetDZFDataDB(db string) { + dzfDataDB = db +} + // Database returns the configured database name. func Database() string { return cfg.Database @@ -154,6 +167,10 @@ func Load() error { dzdpDB = db } + if db := os.Getenv("CLICKHOUSE_DZF_DATA_DB"); db != "" { + dzfDataDB = db + } + // Build env -> database mapping. // Devnet and testnet databases default to lake_devnet / lake_testnet // unless overridden by env vars or disabled with CLICKHOUSE_NO_DEVNET / CLICKHOUSE_NO_TESTNET. @@ -177,7 +194,7 @@ func Load() error { secure := os.Getenv("CLICKHOUSE_SECURE") == "true" - slog.Info("connecting to ClickHouse", "addr", cfg.Addr, "database", cfg.Database, "username", cfg.Username, "secure", secure, "shredder_db", shredderDB, "publisher_db", publisherDB, "dzdp_db", dzdpDB) + slog.Info("connecting to ClickHouse", "addr", cfg.Addr, "database", cfg.Database, "username", cfg.Username, "secure", secure, "shredder_db", shredderDB, "publisher_db", publisherDB, "dzdp_db", dzdpDB, "dzf_data_db", dzfDataDB) // Create connection pool opts := &clickhouse.Options{ diff --git a/api/handlers/api.go b/api/handlers/api.go index 7dcc968a3..c002ef4c1 100644 --- a/api/handlers/api.go +++ b/api/handlers/api.go @@ -43,6 +43,7 @@ type API struct { ShredderDB string PublisherDB string DZDPDB string + DZFDataDB string // PostgreSQL PgPool *pgxpool.Pool diff --git a/api/handlers/publisher_check.go b/api/handlers/publisher_check.go index 32d44d0de..701374e0f 100644 --- a/api/handlers/publisher_check.go +++ b/api/handlers/publisher_check.go @@ -67,6 +67,8 @@ type PublisherCheckItem struct { ValidatorName string `json:"validator_name"` ValidatorVersionOk bool `json:"validator_version_ok"` IsBackup bool `json:"is_backup"` + Lagging bool `json:"lagging"` + LaggingStatus string `json:"lagging_status"` } // PublisherCheckResponse is the response for the publisher check endpoint. @@ -325,6 +327,20 @@ func (a *API) FetchPublisherCheckData(ctx context.Context, q string, epochsParam publishers = []PublisherCheckItem{} } + // Mark publishers whose validator is reported as lagging (status != Healthy) + // in dzf_data.lagged_validators. Non-fatal: if the lookup fails we still + // return publisher data without lagging annotations. + if lagging, err := a.fetchLaggedValidators(ctx); err != nil { + slog.Warn("publisher check: lagged validators query failed", "error", err) + } else { + for i := range publishers { + if status, ok := lagging[publishers[i].VotePubkey]; ok { + publishers[i].Lagging = true + publishers[i].LaggingStatus = status + } + } + } + var totalNetworkStake int64 err = a.envDB(ctx).QueryRow(ctx, `SELECT COALESCE(SUM(activated_stake_lamports), 0) @@ -358,3 +374,34 @@ func (a *API) FetchPublisherCheckData(ctx context.Context, q string, epochsParam Publishers: publishers, }, nil } + +// fetchLaggedValidators returns a map of validator vote pubkey to its lagging +// status for every validator currently reported as not Healthy in +// dzf_data.lagged_validators. +func (a *API) fetchLaggedValidators(ctx context.Context) (map[string]string, error) { + start := time.Now() + query := fmt.Sprintf(` + SELECT validator_vote_public_key, status + FROM `+"`%s`"+`.lagged_validators + WHERE status != 'Healthy' AND validator_vote_public_key != ''`, a.DZFDataDB) + + rows, err := a.envDB(ctx).Query(ctx, query) + metrics.RecordClickHouseQuery("lagged_validators", time.Since(start), err) + if err != nil { + return nil, err + } + defer rows.Close() + + lagging := make(map[string]string) + for rows.Next() { + var votePubkey, status string + if err := rows.Scan(&votePubkey, &status); err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + lagging[votePubkey] = status + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("rows: %w", err) + } + return lagging, nil +} diff --git a/api/main.go b/api/main.go index 1ad08e1c2..dc03085a3 100644 --- a/api/main.go +++ b/api/main.go @@ -362,6 +362,7 @@ func main() { ShredderDB: config.GetShredderDB(), PublisherDB: config.GetPublisherDB(), DZDPDB: config.GetDZDPDB(), + DZFDataDB: config.GetDZFDataDB(), PgPool: config.PgPool, Neo4jClient: config.Neo4jClient, Neo4jDatabase: config.Neo4jDatabase, diff --git a/scripts/setup-remote-tables.sh b/scripts/setup-remote-tables.sh index 5ebbfc931..05726ce10 100755 --- a/scripts/setup-remote-tables.sh +++ b/scripts/setup-remote-tables.sh @@ -178,6 +178,9 @@ echo "" EXTERNAL_TABLES=( "shredder:publisher_shred_stats" "shredder_qa:publisher_shred_stats" + # lagged_validators is filled by an external process and only exists in prod; + # proxy just this table (validator_rewards is ingested locally by the indexer). + "dzf_data:lagged_validators" ) if [[ ${#EXTERNAL_TABLES[@]} -gt 0 ]]; then diff --git a/web/src/components/publisher-check-page.tsx b/web/src/components/publisher-check-page.tsx index 4251315b1..8f3503b62 100644 --- a/web/src/components/publisher-check-page.tsx +++ b/web/src/components/publisher-check-page.tsx @@ -33,7 +33,8 @@ type SortField = | "publishing_leader_shreds" | "publishing_retransmitted" | "leader_slots" - | "validator_client"; + | "validator_client" + | "lagging"; type SortDirection = "asc" | "desc"; @@ -412,6 +413,9 @@ export function PublisherCheckPage() { `${b.validator_client} ${b.validator_version}`, ); break; + case "lagging": + cmp = Number(a.lagging) - Number(b.lagging); + break; default: cmp = 0; } @@ -867,6 +871,10 @@ export function PublisherCheckPage() { No Retransmit Shreds + handleSort("lagging")}> + Fast Shreds + + handleSort("leader_slots")} @@ -887,7 +895,7 @@ export function PublisherCheckPage() { {pagedPublishers.length === 0 ? ( {activeFilter @@ -974,6 +982,11 @@ export function PublisherCheckPage() { + + + {pub.leader_slots.toLocaleString()} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index bad20bb17..fbcf42e78 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -5908,6 +5908,8 @@ export interface PublisherCheckItem { validator_name: string validator_version_ok: boolean is_backup: boolean + lagging: boolean + lagging_status: string } export interface PublisherCheckResponse {