From 2b53fe6e471d057f55c0515540ebc0f116bd92a5 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 3 Aug 2026 13:24:42 -0700 Subject: [PATCH] feat(validators): give the nominator-counts lane a D1 sink so it can advance again `nominator_count` on /api/v1/validators and /api/v1/validators/{hotkey} is served from the lakehouse mirror of `chain.validator_nominator_counts` (#9276), and that mirror is frozen. Measured live 2026-08-03: 112,550 rows, newest `captured_at` 2026-08-02T01:38Z -- the retired box's last scan -- covering only 564 of the 1,031 validators the leaderboard serves. Coverage can only fall from here as new validators register. This is a SINK problem, not a producer problem. metagraphed-infra's poller Container already carries the full SubtensorModule::Alpha scan this table is derived from (src/bin/poller/jobs/validator_nominators.rs, 24h tick); it is one of the five lanes Dockerfile.poller leaves disabled "until they have a Cloudflare-native sink", because it writes to a Postgres that no longer exists. The scan needs no rehosting. Its write target does. The lane, on the pattern #9273/#9292 used for `nominator_positions` -- the other output of that same scan: - migrations/d1/0012 adds `validator_nominator_counts` on D1, upserted on (hotkey) with the usual `captured_at <= excluded.captured_at` staleness guard. The writer's column list and the table are asserted against each other in both directions, the same anti-drift guarantee 0007 has. No prune, and for a sharper reason than account_identity's: the producer's pass over Alpha IS exhaustive, but it chunks across requests, so "absent from this batch" and "absent from the scan" are different statements and only the latter would license a delete. - src/validator-nominator-counts-d1-write.ts reuses neurons-d1-write's statement building, so one place owns the Workers-binding budget of 100 bound parameters per statement -- asserted as the platform limit in the tests, not as a constant we picked. At 3 columns this chunks to 30 rows a statement, so a full scan is ~3,752 statements. - workers/data-api.ts's handleValidatorNominatorCountsSync stops answering `503 hyperdrive binding unavailable` and writes that batch to D1. `nominator_count` is required to be a non-negative integer rather than coerced: the read side already discards anything else, and a value the route accepted but every reader silently drops is worse than a 400 the producer can see. - The read is wired into buildGlobalValidators/buildValidatorDetail, which were handed a hardcoded empty map / null count. It joins against `neurons` inside SQLite rather than inlining a key list -- the leaderboard covers ~1,031 hotkeys against a 100-parameter cap, so a correlated subquery costs zero bound parameters and one query where an IN list would cost a dozen round trips. #9276's serving-Worker lakehouse overlay is deliberately left in place. It only collects hotkeys whose count is still null and returns early when there are none, so it covers what D1 has not received yet and stops firing on its own once a full scan lands -- the cutover is a property of the data, not of a deploy. Failure posture is the family's throughout: any failed read degrades to null, which is exactly what this tier served before the table existed. A broken read is a lost enrichment, never a wrong number. Closes #9301 Part of #9146 --- .../d1/0012_validator_nominator_counts.sql | 47 +++ src/validator-nominator-counts-d1-write.ts | 55 ++++ tests/data-api-neurons-d1.test.ts | 122 ++++++++ ...-api-validator-nominator-counts-d1.test.ts | 240 +++++++++++++++ tests/data-api.test.ts | 8 +- ...alidator-nominator-counts-d1-write.test.ts | 158 ++++++++++ workers/data-api.ts | 280 +++++++++++++++--- 7 files changed, 868 insertions(+), 42 deletions(-) create mode 100644 migrations/d1/0012_validator_nominator_counts.sql create mode 100644 src/validator-nominator-counts-d1-write.ts create mode 100644 tests/data-api-validator-nominator-counts-d1.test.ts create mode 100644 tests/validator-nominator-counts-d1-write.test.ts diff --git a/migrations/d1/0012_validator_nominator_counts.sql b/migrations/d1/0012_validator_nominator_counts.sql new file mode 100644 index 0000000000..077b522fd6 --- /dev/null +++ b/migrations/d1/0012_validator_nominator_counts.sql @@ -0,0 +1,47 @@ +-- The validator-nominator-counts family on D1 (box decommission, the +-- #9146/#9157 pattern applied to the last frozen sync lane). +-- +-- WHY THIS LANE WAS DEAD RATHER THAN MERELY MIGRATED. Its producer -- a full +-- SubtensorModule::Alpha scan -- was already ported off the box into the +-- poller Container (metagraphed-infra's +-- src/bin/poller/jobs/validator_nominators.rs), but that job writes straight +-- to Postgres, so it is one of the five lanes Dockerfile.poller leaves +-- disabled "until they have a Cloudflare-native sink". This table is that +-- sink. The scan itself needs no rehosting; only its write target does. +-- +-- Latest-only, upserted on (hotkey) with the same +-- `captured_at <= excluded.captured_at` staleness guard every other D1 sync +-- lane uses. NO history table: a nominator count is a live gauge, not a fact +-- worth diffing over time (handleValidatorNominatorCountsSync's own header +-- made this call, and nothing about the D1 port changes it). +-- +-- NO PRUNE, deliberately -- and for a DIFFERENT reason than account_identity's +-- in 0009. There, a missing account might simply not have been observed. Here +-- the producer's pass over Alpha is exhaustive by construction, so a hotkey +-- absent from a batch genuinely has no stake entries at all. That would argue +-- FOR a prune, except the producer chunks its batches (the sync route caps +-- rows per request), so "absent from this batch" and "absent from the scan" +-- are not the same statement and only the latter licenses a delete. Upsert-only +-- also matches what handleValidatorNominatorCountsSync did against Postgres, +-- so the port changes the store and nothing else. +-- +-- Type translation follows 0007_neurons.sql's conventions: counts -> INTEGER, +-- captured_at -> INTEGER epoch-ms, hotkey (SS58) -> TEXT. +-- +-- The column set is NOT transcribed by hand: it is exactly the list the writer +-- binds, VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS +-- (src/validator-nominator-summary.ts). +-- tests/validator-nominator-counts-d1-write.test.ts asserts that +-- correspondence in both directions -- the same anti-drift guarantee as 0007's +-- tests/neurons-d1-schema.test.ts and 0011's own. +-- +-- SIZE: 112,550 rows live-measured against the lakehouse mirror of this same +-- table (2026-08-03), one row per distinct hotkey ever seen holding stake -- +-- far wider than the ~1,031 hotkeys that currently carry a validator permit, +-- because every nominated hotkey is scanned, not just permitted ones. +CREATE TABLE IF NOT EXISTS validator_nominator_counts ( + hotkey TEXT NOT NULL, + nominator_count INTEGER NOT NULL, + captured_at INTEGER NOT NULL, + PRIMARY KEY (hotkey) +); diff --git a/src/validator-nominator-counts-d1-write.ts b/src/validator-nominator-counts-d1-write.ts new file mode 100644 index 0000000000..72132912cd --- /dev/null +++ b/src/validator-nominator-counts-d1-write.ts @@ -0,0 +1,55 @@ +// The validator-nominator-counts sync write path, against D1 (#9146). +// +// The last of the five Postgres-backed poller lanes to get a Cloudflare-native +// sink. Its producer (metagraphed-infra's +// src/bin/poller/jobs/validator_nominators.rs) already runs in the poller +// Container and already performs the full SubtensorModule::Alpha scan this +// table is derived from -- it is disabled purely because it writes to a +// Postgres that no longer exists. This module is the store it writes to +// instead; migrations/d1/0011_validator_nominator_counts.sql is the table. +// +// Everything structural is imported from src/neurons-d1-write.ts rather than +// re-derived -- D1_PARAM_BUDGET, chunkStatements, batchInSlices -- so the +// binding's 100-bound-parameter limit is enforced in exactly one place. That +// limit is not theoretical here: at 3 columns this table chunks to 30 rows a +// statement, and a full 112,550-row scan is ~3,752 statements, so a +// hand-rolled batch would have hit the same wall #9157 hit in production. +// +// buildUpsert's trailing `captured_at <= excluded.captured_at` guard is +// exactly right for this table and is why chunkStatements is reused verbatim +// instead of forked: the producer chunks one scan across several requests and +// re-sends on failure, so an out-of-order or replayed batch must be a no-op +// rather than a regression to an older count. + +import { + batchInSlices, + chunkStatements, + type D1Like, + type D1PreparedStatement, +} from "./neurons-d1-write.ts"; +import { VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS } from "./validator-nominator-summary.ts"; + +type Row = Record; + +/** + * Write one batch of nominator counts to D1: a latest-only upsert on + * (hotkey), nothing else. + * + * NO PRUNE, and no history append -- see the migration's header for both. An + * empty batch issues no statements at all rather than an empty `db.batch([])`, + * matching writeAccountIdentityToD1's own guard. + */ +export async function writeValidatorNominatorCountsToD1( + db: D1Like, + rows: Row[], +): Promise<{ statements: number }> { + const statements: D1PreparedStatement[] = chunkStatements( + db, + "validator_nominator_counts", + VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS, + ["hotkey"], + rows, + ); + if (statements.length) await batchInSlices(db, statements); + return { statements: statements.length }; +} diff --git a/tests/data-api-neurons-d1.test.ts b/tests/data-api-neurons-d1.test.ts index 7c052bb0d4..74a5dc41c9 100644 --- a/tests/data-api-neurons-d1.test.ts +++ b/tests/data-api-neurons-d1.test.ts @@ -38,6 +38,13 @@ const OBSERVATIONS_SCHEMA = fs.readFileSync( path.join(process.cwd(), "migrations/d1/0002_observations.sql"), "utf8", ); +// validator_nominator_counts (the nominator_count join target, #9146) -- the +// same reason as subnet_snapshots above: the real database carries it, and the +// leaderboard's read joins against it. +const NOMINATOR_COUNTS_SCHEMA = fs.readFileSync( + path.join(process.cwd(), "migrations/d1/0012_validator_nominator_counts.sql"), + "utf8", +); let db: InstanceType; @@ -249,6 +256,7 @@ beforeEach(() => { db.exec(NEURONS_SCHEMA); db.exec(NEURONS_READ_INDEXES); db.exec(OBSERVATIONS_SCHEMA); + db.exec(NOMINATOR_COUNTS_SCHEMA); }); // --- POST /api/v1/internal/neurons-sync: the D1 write lane ------------------- @@ -636,6 +644,120 @@ test("GET /api/v1/validators/:hotkey aggregates one hotkey across subnets with p assert.equal(body.total_stake_tao, 150); }); +// --- nominator_count, joined from D1 (#9146) --------------------------------- +// +// This field was null on EVERY validator for the whole period between the box +// wipe and migration 0012: the side table was Postgres-only, so both builders +// were handed a hardcoded empty map / null. What is under test is that the +// join now answers, and -- just as important -- that a hotkey with no row +// still reads as UNKNOWN rather than as a confident zero. + +/** One row in the counts side table. */ +function insertNominatorCount(hotkey: string, count: number, at = 1_000) { + db.prepare( + "INSERT INTO validator_nominator_counts (hotkey, nominator_count, captured_at) VALUES (?, ?, ?)", + ).run(hotkey, count, at); +} + +test("GET /api/v1/validators fills nominator_count from D1, and leaves an unscanned hotkey null", async () => { + insertNeuron({ + netuid: 0, + uid: 0, + hotkey: "5Counted", + stake_tao: 200, + validator_permit: 1, + }); + insertNeuron({ + netuid: 0, + uid: 1, + hotkey: "5Unscanned", + stake_tao: 100, + validator_permit: 1, + }); + insertNominatorCount("5Counted", 42); + + const res = await call(req("/api/v1/validators")); + assert.equal(res.status, 200); + const entries = ((await res.json()) as Row).validators as Row[]; + const byHotkey = new Map(entries.map((e) => [e.hotkey as string, e])); + + assert.equal(byHotkey.get("5Counted")!.nominator_count, 42); + assert.equal( + byHotkey.get("5Unscanned")!.nominator_count, + null, + "a hotkey the scan has not reached is unknown, NOT a confident zero", + ); +}); + +test("GET /api/v1/validators/:hotkey fills nominator_count for that one hotkey", async () => { + insertNeuron({ + netuid: 0, + uid: 0, + hotkey: "5Val", + stake_tao: 100, + validator_permit: 1, + }); + insertNominatorCount("5Val", 7); + // A second hotkey's row must not leak into this one's answer. + insertNominatorCount("5Other", 999); + + const res = await call(req("/api/v1/validators/5Val")); + assert.equal(res.status, 200); + assert.equal(((await res.json()) as Row).nominator_count, 7); +}); + +test("GET /api/v1/validators/:hotkey reports a zero count as a real answer", async () => { + // 0 is an ANSWER, not an absence -- the one case where the distinction the + // rest of this lane preserves has to survive all the way to the payload. + insertNeuron({ + netuid: 0, + uid: 0, + hotkey: "5Val", + stake_tao: 100, + validator_permit: 1, + }); + insertNominatorCount("5Val", 0); + const res = await call(req("/api/v1/validators/5Val")); + assert.equal(((await res.json()) as Row).nominator_count, 0); +}); + +test("GET /api/v1/validators/:hotkey leaves nominator_count null when the hotkey has no row", async () => { + insertNeuron({ + netuid: 0, + uid: 0, + hotkey: "5Val", + stake_tao: 100, + validator_permit: 1, + }); + const res = await call(req("/api/v1/validators/5Val")); + assert.equal(((await res.json()) as Row).nominator_count, null); +}); + +test("a failing counts read degrades to null rather than failing the request", async () => { + // The whole lane's failure posture: the leaderboard is the product, the + // count is an enrichment. Dropping the table is the bluntest way to make the + // read throw for real, rather than asserting a mocked rejection. + insertNeuron({ + netuid: 0, + uid: 0, + hotkey: "5Val", + stake_tao: 100, + validator_permit: 1, + }); + db.exec("DROP TABLE validator_nominator_counts"); + + const list = await call(req("/api/v1/validators")); + assert.equal(list.status, 200, "the leaderboard still serves"); + assert.equal( + (((await list.json()) as Row).validators as Row[])[0]!.nominator_count, + null, + ); + + const detail = await call(req("/api/v1/validators/5Val")); + assert.equal(detail.status, 200, "the detail card still serves"); + assert.equal(((await detail.json()) as Row).nominator_count, null); +}); + // --- Live-neurons analytics routes ------------------------------------------- test("GET /api/v1/subnets/:netuid/concentration and /performance read the D1 snapshot", async () => { diff --git a/tests/data-api-validator-nominator-counts-d1.test.ts b/tests/data-api-validator-nominator-counts-d1.test.ts new file mode 100644 index 0000000000..1bbd1090ff --- /dev/null +++ b/tests/data-api-validator-nominator-counts-d1.test.ts @@ -0,0 +1,240 @@ +// The revived validator-nominator-counts sync lane (#9146), exercised END TO +// END against a REAL SQLite database through the real Worker fetch handler -- +// same harness and rationale as tests/data-api-nominator-positions-d1.test.ts. +// +// This route answered `503 hyperdrive binding unavailable` from the box wipe +// (#9193) until migration 0012 gave it a Cloudflare-native store. Its producer +// never went away: metagraphed-infra's poller Container already runs the full +// SubtensorModule::Alpha scan this table is derived from, and was disabled only +// because it wrote to a Postgres that no longer exists. +// +// What matters here is the write CONTRACT. A full scan is ~113k rows, past any +// single request body, so it arrives across SEVERAL requests -- which is why +// there is no prune on this lane at all, and why the staleness guard rather +// than request ordering is what keeps a replay safe. +import assert from "node:assert/strict"; +import { DatabaseSync } from "node:sqlite"; +import fs from "node:fs"; +import path from "node:path"; +import { beforeEach, describe, test } from "vitest"; +import type { Row } from "./row-type.ts"; + +const { default: worker } = await import("../workers/data-api.ts"); + +const SCHEMA = fs.readFileSync( + path.join(process.cwd(), "migrations/d1/0012_validator_nominator_counts.sql"), + "utf8", +); + +const PATH = "/api/v1/internal/validator-nominator-counts-sync"; +const SECRET = "test-validator-nominator-counts-sync-secret"; +const HOTKEY = "5FyVinYphF6JS5FZHzhMQffxtgbz1WxwUEBAxTRo9nABwb5g"; +const HOTKEY_B = "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"; + +let db: InstanceType; + +function d1() { + return { + prepare(text: string) { + return { + bind(...values: unknown[]) { + return { + text, + values, + async all() { + return { results: db.prepare(text).all(...(values as never[])) }; + }, + }; + }, + }; + }, + async batch(statements: { text: string; values: unknown[] }[]) { + db.exec("BEGIN"); + try { + const results = statements.map((statement) => ({ + results: db + .prepare(statement.text) + .all(...(statement.values as never[])), + })); + db.exec("COMMIT"); + return results; + } catch (err) { + db.exec("ROLLBACK"); + throw err; + } + }, + }; +} + +function env(overrides: Record = {}): Env { + return { + METAGRAPH_HEALTH_DB: d1(), + VALIDATOR_NOMINATOR_COUNTS_SYNC_SECRET: SECRET, + ...overrides, + } as unknown as Env; +} + +function post(body: unknown, token: string | null = SECRET, envOverride?: Env) { + const headers: Record = { + "content-type": "application/json", + }; + if (token !== null) + headers["x-validator-nominator-counts-sync-token"] = token; + return worker.fetch( + new Request(`https://d${PATH}`, { + method: "POST", + headers, + body: typeof body === "string" ? body : JSON.stringify(body), + }), + envOverride ?? env(), + {} as unknown as ExecutionContext, + ); +} + +function countRow(overrides: Row = {}): Row { + return { + hotkey: HOTKEY, + nominator_count: 12, + captured_at: 1_780_000_000_000, + ...overrides, + }; +} + +const rows = () => + db + .prepare("SELECT * FROM validator_nominator_counts ORDER BY hotkey") + .all() as Row[]; + +beforeEach(() => { + db = new DatabaseSync(":memory:"); + db.exec(SCHEMA); +}); + +describe("POST /api/v1/internal/validator-nominator-counts-sync", () => { + test("writes a batch to D1 and reports what it did", async () => { + const response = await post({ + rows: [countRow(), countRow({ hotkey: HOTKEY_B, nominator_count: 0 })], + }); + assert.equal(response.status, 200); + assert.deepEqual(await response.json(), { + ok: true, + nominator_counts_written: 2, + stores: ["d1"], + d1_statements: 1, + }); + assert.equal(rows().length, 2); + // A zero IS stored -- the producer's scan is exhaustive, so "this hotkey + // has no nominators" is an answer it can legitimately report. + assert.equal(rows()[1]!.nominator_count, 0); + }); + + test("accepts a bare array as well as {rows:[...]}", async () => { + // The producer posts a bare array; every other sync route here speaks + // {rows:[...]}. A mismatch must not cost a whole 24h cycle. + const response = await post([countRow()]); + assert.equal(response.status, 200); + assert.equal(rows().length, 1); + }); + + test("a later capture wins and an older one is a no-op", async () => { + // The staleness guard is what makes a replayed or out-of-order batch safe. + // It matters more on this lane than most: the producer chunks one scan + // across several requests and re-sends on failure. + await post({ rows: [countRow({ nominator_count: 12 })] }); + await post({ + rows: [countRow({ nominator_count: 30, captured_at: 1_780_000_100_000 })], + }); + assert.equal(rows()[0]!.nominator_count, 30); + + await post({ + rows: [countRow({ nominator_count: 1, captured_at: 1_779_000_000_000 })], + }); + assert.equal( + rows()[0]!.nominator_count, + 30, + "an older capture must never walk a count backwards", + ); + }); + + test("rejects a missing or wrong token (401)", async () => { + assert.equal((await post({ rows: [countRow()] }, null)).status, 401); + assert.equal((await post({ rows: [countRow()] }, "nope")).status, 401); + assert.equal(rows().length, 0); + }); + + test("is disabled (503) when the secret is not configured", async () => { + const response = await post( + { rows: [countRow()] }, + SECRET, + env({ VALIDATOR_NOMINATOR_COUNTS_SYNC_SECRET: undefined }), + ); + assert.equal(response.status, 503); + }); + + test("answers 503 when D1 is not bound -- but only after validating (400 wins)", async () => { + // A malformed body is a 400 whether or not a store happens to be bound; + // answering 503 would blame the infrastructure for the caller's payload. + const unbound = env({ METAGRAPH_HEALTH_DB: undefined }); + assert.equal( + (await post({ rows: [countRow()] }, SECRET, unbound)).status, + 503, + ); + assert.equal( + (await post({ rows: [countRow({ hotkey: 1 })] }, SECRET, unbound)).status, + 400, + "validation runs before the binding check", + ); + }); + + test("rejects a body that is not JSON, or not a row array", async () => { + assert.equal((await post("not json")).status, 400); + assert.equal((await post({ rows: "nope" })).status, 400); + assert.equal((await post({ rows: [] })).status, 400); + }); + + test("rejects rows that do not match the column shape", async () => { + const bad: Row[] = [ + { ...countRow(), unexpected: 1 }, + countRow({ hotkey: "" }), + countRow({ hotkey: 5 }), + countRow({ hotkey: "x".repeat(200) }), + countRow({ nominator_count: -1 }), + countRow({ nominator_count: 1.5 }), + countRow({ nominator_count: "12" }), + countRow({ captured_at: 0 }), + countRow({ captured_at: 1.5 }), + "not an object" as unknown as Row, + null as unknown as Row, + [] as unknown as Row, + ]; + for (const row of bad) { + const response = await post({ rows: [row] }); + assert.equal( + response.status, + 400, + `expected 400 for ${JSON.stringify(row)}`, + ); + } + assert.equal(rows().length, 0, "no partial write from a rejected batch"); + }); + + test("rejects an oversized batch (413) by rows and by bytes", async () => { + const tooMany = Array.from({ length: 50_001 }, (_unused, i) => + countRow({ hotkey: `hk-${i}` }), + ); + assert.equal((await post({ rows: tooMany })).status, 413); + + // Body bound is checked before a row count exists -- a handful of enormous + // strings passes the row bound and must still be refused. + const huge = JSON.stringify({ rows: [countRow()] }).padEnd(8_000_001, " "); + assert.equal((await post(huge)).status, 413); + assert.equal(rows().length, 0); + }); + + test("a D1 failure is a 502, not a silent success", async () => { + db.exec("DROP TABLE validator_nominator_counts"); + const response = await post({ rows: [countRow()] }); + assert.equal(response.status, 502); + assert.deepEqual(await response.json(), { error: "d1 write failed" }); + }); +}); diff --git a/tests/data-api.test.ts b/tests/data-api.test.ts index 129c7255f4..f57f7b125a 100644 --- a/tests/data-api.test.ts +++ b/tests/data-api.test.ts @@ -1068,7 +1068,12 @@ test("validator-nominator-counts-sync is disabled (503) when VALIDATOR_NOMINATOR expect(res.status).toBe(503); }); -test("validator-nominator-counts-sync answers 503 -- the Postgres tier it wrote to is gone (#9193)", async () => { +test("validator-nominator-counts-sync answers 503 when no store is bound (#9146)", async () => { + // Was "the Postgres tier it wrote to is gone (#9193)". The lane is no longer + // retired -- migration 0012 gave it a D1 store and the handler writes there + // (tests/data-api-validator-nominator-counts-d1.test.ts covers the write + // against a real database). What this env pins is the remaining 503: an + // authenticated, well-formed request with NOTHING bound to write to. const res = await worker.fetch( new Request("https://d/api/v1/internal/validator-nominator-counts-sync", { method: "POST", @@ -1083,6 +1088,7 @@ test("validator-nominator-counts-sync answers 503 -- the Postgres tier it wrote ctx, ); expect(res.status).toBe(503); + expect(await res.json()).toEqual({ error: "d1 binding unavailable" }); }); // #5233: POST /api/v1/internal/nominator-positions-sync -- the write path diff --git a/tests/validator-nominator-counts-d1-write.test.ts b/tests/validator-nominator-counts-d1-write.test.ts new file mode 100644 index 0000000000..73909288cc --- /dev/null +++ b/tests/validator-nominator-counts-d1-write.test.ts @@ -0,0 +1,158 @@ +// The validator-nominator-counts D1 write path (#9146) and the schema it +// writes into. +// +// Two things are checked here and they fail in different ways. The MIGRATION +// check is anti-drift: a column the writer binds but the table lacks makes D1 +// reject the whole batch, and a column the table has but the writer never +// sends is a permanently-NULL field that reads like real data (0007's +// tests/neurons-d1-schema.test.ts and 0011's +// tests/nominator-positions-d1-write.test.ts make the same guarantee for their +// own tables). The WRITER checks are about the parameter budget and the +// staleness guard: this lane posts a ~113k-row scan across several requests, +// so a chunk that overran the binding's limit would fail the whole batch, and +// a replayed request must never walk a count backwards. +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { describe, test } from "vitest"; +import { writeValidatorNominatorCountsToD1 } from "../src/validator-nominator-counts-d1-write.ts"; +import { VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS } from "../src/validator-nominator-summary.ts"; +import { D1_PARAM_BUDGET } from "../src/neurons-d1-write.ts"; + +const MIGRATION = readFileSync( + "migrations/d1/0012_validator_nominator_counts.sql", + "utf8", +); + +const HOTKEY = "5FyVinYphF6JS5FZHzhMQffxtgbz1WxwUEBAxTRo9nABwb5g"; + +/** Column names of one CREATE TABLE block, in declaration order. */ +function tableColumns(table: string): string[] { + const match = MIGRATION.match( + new RegExp(`CREATE TABLE IF NOT EXISTS ${table} \\(([\\s\\S]*?)\\n\\);`), + ); + assert.ok(match, `no CREATE TABLE for ${table} in the migration`); + return match[1] + .split("\n") + .map((line) => line.trim()) + .filter( + (line) => + line && + !line.startsWith("--") && + !line.startsWith("PRIMARY KEY") && + !line.startsWith("CHECK"), + ) + .map((line) => line.split(/\s+/)[0]); +} + +/** Records every prepared statement and its bindings, in order. */ +function d1Stub() { + const statements: { sql: string; params: unknown[] }[] = []; + const batches: number[] = []; + const db = { + prepare(sql: string) { + return { + bind(...params: unknown[]) { + const entry = { sql, params }; + statements.push(entry); + return entry as never; + }, + }; + }, + async batch(slice: unknown[]) { + batches.push(slice.length); + return []; + }, + }; + return { statements, batches, db }; +} + +function row(hotkey: string, count: number, at: number) { + return { hotkey, nominator_count: count, captured_at: at }; +} + +describe("the validator_nominator_counts D1 schema matches its writer", () => { + test("the migration parser actually finds columns", () => { + // A regex that silently matched nothing would make every comparison below + // vacuously pass -- the way a source-scanning check stops checking. + assert.equal(tableColumns("validator_nominator_counts").length, 3); + }); + + test("the table holds exactly the columns the sync binds", () => { + assert.deepEqual( + tableColumns("validator_nominator_counts").sort(), + [...VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS].sort(), + ); + }); + + test("the upsert key is one row per hotkey", () => { + // Latest-only: the Postgres original was PRIMARY KEY (hotkey) with + // REPLACE-on-conflict, and the serving readers assume one row per hotkey + // (the lakehouse mirror, which does NOT enforce this, needs a group-wise + // MAX for exactly that reason -- see + // src/validator-nominator-counts-cold-tier.ts). + assert.match( + MIGRATION, + /PRIMARY KEY \(hotkey\)/, + "the PRIMARY KEY must be the same key the writer declares as its conflict target", + ); + }); +}); + +describe("writeValidatorNominatorCountsToD1", () => { + test("upserts on hotkey with the staleness guard", async () => { + const { statements, db } = d1Stub(); + const { statements: count } = await writeValidatorNominatorCountsToD1( + db as never, + [row(HOTKEY, 12, 1_000), row("5G9", 3, 1_000)], + ); + + assert.equal(count, statements.length); + assert.equal(statements.length, 1, "two narrow rows fit one statement"); + assert.match( + statements[0]!.sql, + /INSERT INTO validator_nominator_counts \(hotkey, nominator_count, captured_at\)/, + ); + assert.match(statements[0]!.sql, /ON CONFLICT \(hotkey\) DO UPDATE SET/); + assert.match( + statements[0]!.sql, + /WHERE validator_nominator_counts\.captured_at <= excluded\.captured_at/, + "an older capture must never overwrite a newer one", + ); + assert.deepEqual(statements[0]!.params, [ + HOTKEY, + 12, + 1_000, + "5G9", + 3, + 1_000, + ]); + }); + + test("no statement exceeds the Workers binding's bound-parameter limit", async () => { + // 100 per statement on the BINDING -- not the 1,200 `wrangler d1 execute` + // permits from the CLI. The first 15 production neurons syncs all failed + // on exactly this, so the limit is asserted, never a constant we picked. + const { statements, db } = d1Stub(); + await writeValidatorNominatorCountsToD1( + db as never, + Array.from({ length: 500 }, (_unused, i) => row(`hk-${i}`, i, 1_000)), + ); + assert.ok(statements.length > 1, "500 rows must chunk"); + for (const statement of statements) { + assert.ok( + statement.params.length <= D1_PARAM_BUDGET, + `a statement bound ${statement.params.length} parameters, over the ${D1_PARAM_BUDGET} budget`, + ); + } + }); + + test("an empty batch issues no statements and never calls batch()", async () => { + // An empty `db.batch([])` is a round trip that can only fail; the producer + // legitimately posts nothing when a chunk boundary lands cleanly. + const { statements, batches, db } = d1Stub(); + const result = await writeValidatorNominatorCountsToD1(db as never, []); + assert.equal(result.statements, 0); + assert.equal(statements.length, 0); + assert.deepEqual(batches, []); + }); +}); diff --git a/workers/data-api.ts b/workers/data-api.ts index e3b4f2a264..ce1860cb04 100644 --- a/workers/data-api.ts +++ b/workers/data-api.ts @@ -109,7 +109,10 @@ import { IDENTITY_FIELDS, buildAccountIdentity, } from "../src/account-identity.ts"; -import {} from "../src/validator-nominator-summary.ts"; +import { + nominatorCountsByHotkey, + VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS, +} from "../src/validator-nominator-summary.ts"; import { NOMINATOR_POSITION_INSERT_COLUMNS } from "../src/account-nominator-positions.ts"; import { identityHash, @@ -302,6 +305,7 @@ import { coldkeyMaxCapturedAt, writeNominatorPositionsToD1, } from "../src/nominator-positions-d1-write.ts"; +import { writeValidatorNominatorCountsToD1 } from "../src/validator-nominator-counts-d1-write.ts"; import { writeChainDetailToD1 } from "../src/chain-detail-d1-write.ts"; import { CHAIN_DETAIL_SYNC_MAX_BODY_BYTES, @@ -1453,21 +1457,72 @@ async function handleAccountIdentitySync(request: Request, env: Env) { // --- POST /api/v1/internal/validator-nominator-counts-sync (#2549) -------- // -// RETIRED (#9193): the Postgres tables this wrote were destroyed with the -// box, so the handler now stops at its auth gate and answers exactly what it -// already answered in production. What follows describes what it DID. +// RESTORED ON D1 (#9146). Retired by #9193 when the Postgres table it wrote +// was destroyed with the box; this is the same route against +// migrations/d1/0011_validator_nominator_counts.sql instead. // -// The write path into validator_nominator_counts (migration 0043) -- -// simpler than account-identity-sync above: latest-only, no history table -// (a nominator count is a live gauge, not a fact worth diffing over time -// yet). Populated by its own low-frequency job -// (apps/indexer-rs/src/bin/poller/jobs/validator_nominators.rs), decoupled -// from the fast refresh-metagraph cron -- see that job's and the migration's +// The write path into validator_nominator_counts -- simpler than +// account-identity-sync above: latest-only, no history table (a nominator +// count is a live gauge, not a fact worth diffing over time yet). Populated by +// its own low-frequency job +// (metagraphed-infra src/bin/poller/jobs/validator_nominators.rs, 24h), +// decoupled from the fast neurons sync -- see that job's and the migration's // own header comments for why a full SubtensorModule::Alpha scan can't share // the neurons snapshot's cadence. +// +// THE PRODUCER CHUNKS; THIS ROUTE DOES NOT REASSEMBLE. One scan is ~112,550 +// rows, over the per-request cap below, so a scan arrives as several +// independent requests. That is why there is no prune here and no batch-wide +// bookkeeping: each request is a self-contained upsert, and correctness comes +// from buildUpsert's captured_at guard rather than from requests arriving in +// order or at all. A dropped chunk costs those hotkeys one cycle of freshness, +// never a wrong value. const VALIDATOR_NOMINATOR_COUNTS_SYNC_TOKEN_HEADER = "x-validator-nominator-counts-sync-token"; +// 50k rows x 3 narrow columns is ~4 MB, putting a full ~113k-row scan at 3 +// requests. Body bound first, row bound second -- neither alone is sufficient, +// the same pairing nominator-positions-sync above spells out. +const VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_BODY_BYTES = 8_000_000; +const VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_ROWS = 50_000; +// Same 128-byte ceiling nominator-positions-sync puts on its SS58 keys, for +// the same reason: an address is 48 characters, and the slack is so a future +// address format does not silently fail the whole batch. +const VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_KEY_BYTES = 128; + +/** + * Bounds-check one incoming row against + * VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS. + * + * `nominator_count` is required to be a non-negative integer rather than + * coerced into one, because the read side (nominatorCountsByHotkey) already + * discards anything that isn't -- a value this route accepted but every reader + * silently drops is worse than a 400 the producer can actually see. + */ +function validNominatorCountSyncRow(row: Row) { + if (!row || typeof row !== "object" || Array.isArray(row)) return false; + for (const key of Object.keys(row)) { + if (!VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS.includes(key)) return false; + } + if (typeof row.hotkey !== "string" || row.hotkey.length === 0) return false; + if ( + utf8Bytes(row.hotkey).length > VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_KEY_BYTES + ) + return false; + if (!Number.isInteger(row.nominator_count) || row.nominator_count < 0) + return false; + if (!Number.isInteger(row.captured_at) || row.captured_at <= 0) return false; + return true; +} + +/** Project a validated row onto the writer's exact column list and order. */ +function coerceNominatorCountSyncRow(row: Row) { + const out: Row = {}; + for (const col of VALIDATOR_NOMINATOR_COUNT_INSERT_COLUMNS) + out[col] = row[col]; + return out; +} + async function handleValidatorNominatorCountsSync(request: Request, env: Env) { if (!env.VALIDATOR_NOMINATOR_COUNTS_SYNC_SECRET) { return writeJson( @@ -1491,9 +1546,87 @@ async function handleValidatorNominatorCountsSync(request: Request, env: Env) { 401, ); } - // #9193: same deletion as handleRollupAccountEventsDaily above -- unreachable - // since HYPERDRIVE went away, answered here, status and body unchanged. - return writeJson({ error: "hyperdrive binding unavailable" }, 503); + + const raw = await request.text(); + if (utf8Bytes(raw).length > VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_BODY_BYTES) { + return writeJson( + { + error: `body exceeds ${VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_BODY_BYTES} bytes`, + }, + 413, + ); + } + let parsed; + try { + parsed = JSON.parse(raw); + } catch { + return writeJson({ error: "body must be JSON" }, 400); + } + // Both envelopes accepted, matching handleNeuronsSync's own tolerance -- the + // producer sends a bare array, but {rows:[...]} is what every other sync + // route here speaks and a mismatch is not worth a failed cycle. + const incoming = Array.isArray(parsed) + ? parsed + : Array.isArray(parsed?.rows) + ? parsed.rows + : null; + if (!incoming) { + return writeJson( + { + error: + "body must be a JSON array of nominator count rows (or {rows:[...]})", + }, + 400, + ); + } + if (incoming.length > VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_ROWS) { + return writeJson( + { + error: `at most ${VALIDATOR_NOMINATOR_COUNTS_SYNC_MAX_ROWS} rows per request`, + }, + 413, + ); + } + if (!incoming.length || !incoming.every(validNominatorCountSyncRow)) { + return writeJson( + { error: "rows must match the nominator count row shape" }, + 400, + ); + } + + const rows = incoming.map(coerceNominatorCountSyncRow); + + // D1 is the binding this path REQUIRES -- the only store this family has. + // Checked HERE, after validation, not at the top: a malformed body is a 400 + // whether or not a store happens to be bound (handleNominatorPositionsSync's + // own 400-before-503 reasoning). + if (!env.METAGRAPH_HEALTH_DB) { + return writeJson({ error: "d1 binding unavailable" }, 503); + } + + let d1Statements: number; + try { + ({ statements: d1Statements } = await writeValidatorNominatorCountsToD1( + env.METAGRAPH_HEALTH_DB as unknown as Parameters< + typeof writeValidatorNominatorCountsToD1 + >[0], + rows, + )); + } catch (err) { + console.error( + "data-api validator-nominator-counts-sync D1 write failed:", + err, + ); + await captureDataApiError(err, "validator-nominator-counts-sync-d1", env); + return writeJson({ error: "d1 write failed" }, 502); + } + + return writeJson({ + ok: true, + nominator_counts_written: rows.length, + stores: ["d1"], + d1_statements: d1Statements, + }); } // --- POST /api/v1/internal/nominator-positions-sync (#5233, revived #9273) -- @@ -4483,20 +4616,22 @@ async function handleAccountKeysRoute(request: Request, env: Env, url: URL) { // Cross-tier joins: subnet_snapshots has a live D1 home (migrations/d1/ // 0002_observations.sql), so the alpha_price_tao joins/loads port for real. // The remaining enrichment side tables (featured_validators, -// validator_nominator_counts, subnet_hyperparams' tempo/immunity_period, -// account_identity) have NO D1 home yet -- their families are frozen or port -// separately -- so the twins pass each builder the degraded value the retired -// Postgres loader's own catch branch produced (empty set/map, null), rather -// than issuing a query that can only ever throw. Wire the real reads in when -// those tables land on D1. +// subnet_hyperparams' tempo/immunity_period, account_identity) have NO D1 home +// yet -- their families are frozen or port separately -- so the twins pass each +// builder the degraded value the retired Postgres loader's own catch branch +// produced (empty set/map, null), rather than issuing a query that can only +// ever throw. Wire the real reads in when those tables land on D1. // -// One of them is no longer degraded downstream: nominator_count is filled from -// chain.validator_nominator_counts by the SERVING Worker, at tier convergence -// (src/validator-nominator-counts-cold-tier.ts, #9146). It could not be filled -// here -- R2_SQL_TOKEN is bound to the main Worker, not to this one -- so the -// empty map / null count below stay correct as this tier's own answer. If this -// table ever does land on D1, wiring it here makes that overlay a no-op rather -// than a conflict: it only ever fills a count the payload left null. +// validator_nominator_counts NO LONGER BELONGS TO THAT LIST. It landed on D1 +// in migrations/d1/0012, so the real read is wired below and this tier answers +// nominator_count itself. That is exactly the resolution this comment +// anticipated: the serving Worker's lakehouse overlay +// (src/validator-nominator-counts-cold-tier.ts, #9146) becomes a no-op rather +// than a conflict, because validatorHotkeysNeedingCount only collects hotkeys +// whose count is still null and returns early when there are none. The overlay +// stays in place while the producer backfills -- covering hotkeys D1 has not +// received yet from the frozen mirror -- and stops firing on its own once a +// full scan has landed. type NeuronsD1RouteHandler = (sql: D1Sql, env: Env) => Promise; // The D1 twin of loadAlphaPricesByNetuid (#9051): netuid -> latest @@ -4529,6 +4664,65 @@ async function loadAlphaPricesByNetuidD1( } } +// hotkey -> nominator_count for every permitted validator, from the D1 table +// migrations/d1/0012 created (#9146). +// +// CORRELATED SUBQUERY, NOT AN IN LIST, and that is load-bearing rather than +// stylistic: the leaderboard covers ~1,031 hotkeys and the Workers D1 binding +// caps a statement at 100 bound parameters, so an inlined key list would need +// chunking into a dozen round trips (what the lakehouse reader has to do, +// having no join to reach for). Joining against `neurons` inside SQLite costs +// ZERO bound parameters and one query, and keeps the filter exactly in step +// with the leaderboard's own `validator_permit = 1 AND hotkey IS NOT NULL`. +// +// Same degrade-to-empty-map contract as loadAlphaPricesByNetuidD1 above: on +// any failure every nominator_count stays null, which is precisely the state +// this tier served before the table existed -- so a broken read is a lost +// enrichment, never a wrong number. +async function loadNominatorCountsD1( + sql: D1Sql, + env: Env, +): Promise> { + try { + const rows = await sql` + SELECT hotkey, nominator_count FROM validator_nominator_counts + WHERE hotkey IN ( + SELECT DISTINCT hotkey FROM neurons + WHERE validator_permit = 1 AND hotkey IS NOT NULL + )`; + return nominatorCountsByHotkey(rows); + } catch (err) { + console.error("validator_nominator_counts query failed:", err); + await captureDataApiError(err, "validator-nominator-counts-query", env); + return new Map(); + } +} + +// The single-hotkey twin of the above, for /api/v1/validators/{hotkey}. Returns +// null -- not 0 -- when the hotkey has no row, keeping "unknown" and "confirmed +// zero" distinguishable exactly as buildValidatorDetail's own contract requires. +async function loadNominatorCountD1( + sql: D1Sql, + hotkey: string, + env: Env, +): Promise { + try { + const rows = await sql.unsafe( + "SELECT hotkey, nominator_count FROM validator_nominator_counts WHERE hotkey = ?", + [hotkey], + ); + return nominatorCountsByHotkey(rows).get(hotkey) ?? null; + } catch (err) { + console.error("validator_nominator_counts detail query failed:", err); + await captureDataApiError( + err, + "validator-nominator-count-detail-query", + env, + ); + return null; + } +} + // The D1 twin of loadRealizedStakeBaselines (#7228/#9051): per-hotkey // baseline TAO-priced stake ~1d/1w/1m back from neuron_daily. The Postgres // original's `SELECT DISTINCT ON (hotkey) ... ORDER BY hotkey, @@ -4756,14 +4950,16 @@ function matchNeuronsD1Route(url: URL): NeuronsD1RouteHandler | null { limitParam <= GLOBAL_VALIDATOR_LIMIT_MAX ? limitParam : GLOBAL_VALIDATOR_LIMIT_DEFAULT; - const [rows, realizedStakeByHotkey, priceByNetuid] = await Promise.all([ - sql` + const [rows, realizedStakeByHotkey, priceByNetuid, nominatorCounts] = + await Promise.all([ + sql` SELECT netuid, uid, hotkey, coldkey, validator_trust, emission_tao, stake_tao, block_number, captured_at, take FROM neurons WHERE validator_permit = 1 AND hotkey IS NOT NULL ORDER BY hotkey ASC, stake_tao DESC, netuid ASC, uid ASC`, - loadRealizedStakeBaselinesD1(sql, {}, env), - loadAlphaPricesByNetuidD1(sql, env), - ]); + loadRealizedStakeBaselinesD1(sql, {}, env), + loadAlphaPricesByNetuidD1(sql, env), + loadNominatorCountsD1(sql, env), + ]); return json( buildGlobalValidators(rows, { sort, @@ -4771,7 +4967,7 @@ function matchNeuronsD1Route(url: URL): NeuronsD1RouteHandler | null { priceByNetuid, featuredHotkeys: new Set(), identityByColdkey: new Map(), - nominatorCounts: new Map(), + nominatorCounts, tempoByNetuid: new Map(), realizedStakeByHotkey, }), @@ -4786,19 +4982,21 @@ function matchNeuronsD1Route(url: URL): NeuronsD1RouteHandler | null { if (validatorDetail) { return async (sql, env) => { const hotkey = decodeURIComponent(validatorDetail[1]); - const [rows, realizedByHotkey, priceByNetuid] = await Promise.all([ - sql.unsafe( - `SELECT ${NEURON_COLUMNS}, netuid FROM neurons WHERE hotkey = ? AND validator_permit = 1 ORDER BY netuid ASC, uid ASC`, - [hotkey], - ), - loadRealizedStakeBaselinesD1(sql, { hotkey }, env), - loadAlphaPricesByNetuidD1(sql, env), - ]); + const [rows, realizedByHotkey, priceByNetuid, nominatorCount] = + await Promise.all([ + sql.unsafe( + `SELECT ${NEURON_COLUMNS}, netuid FROM neurons WHERE hotkey = ? AND validator_permit = 1 ORDER BY netuid ASC, uid ASC`, + [hotkey], + ), + loadRealizedStakeBaselinesD1(sql, { hotkey }, env), + loadAlphaPricesByNetuidD1(sql, env), + loadNominatorCountD1(sql, hotkey, env), + ]); return json( buildValidatorDetail(rows, hotkey, { identityByColdkey: new Map(), priceByNetuid, - nominatorCount: null, + nominatorCount, tempoByNetuid: new Map(), realizedStake: realizedByHotkey.get(hotkey) ?? null, }),