diff --git a/migrations/d1/0012_validator_nominator_counts.sql b/migrations/d1/0012_validator_nominator_counts.sql new file mode 100644 index 000000000..077b522fd --- /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 000000000..72132912c --- /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 7c052bb0d..74a5dc41c 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 000000000..1bbd1090f --- /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 129c7255f..f57f7b125 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 000000000..73909288c --- /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 e3b4f2a26..ce1860cb0 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, }),