diff --git a/src/sync-batch-queue.ts b/src/sync-batch-queue.ts index bd8d93161..023c476d6 100644 --- a/src/sync-batch-queue.ts +++ b/src/sync-batch-queue.ts @@ -44,7 +44,19 @@ export interface SyncBatchMessage { * it is optional on the sync routes: a producer may post without declaring, * and inventing a total would mark an unproven load complete. */ pass_total?: number; - rows: Record[]; + /** + * The chunk's rows, for a lane that carries one array. + * + * OPTIONAL, because a multi-family message carries `families` INSTEAD and + * `packMultiFamilyMessage` deletes this key outright. It was declared + * required when every lane had rows, and stayed required after `families` + * arrived -- so `message.rows.length` typechecked on a message that has no + * `rows` at all, which is exactly the read that crashed the consumer's batch + * log. The two shapes are mutually exclusive (`validSyncBatchMessage` + * enforces it), so the honest type is "one or the other" and every read goes + * through `syncBatchRows` / `syncBatchRowCount`. + */ + rows?: Record[]; /** * Several row families that must land in ONE write (metagraphed-infra#359). * @@ -101,6 +113,18 @@ export interface SyncBatchMessage { key_value?: string; } +/** + * A message that definitely carries rows, which is what the packer emits. + * + * `rows` is optional on the wire type because the multi-family shape omits it, + * but `packSyncBatchMessages` builds every message from a row array and can + * never produce one without. Narrowing here keeps that guarantee in the type + * system rather than in the reader's memory. + */ +export type SyncBatchRowsMessage = SyncBatchMessage & { + rows: Record[]; +}; + /** * Lanes the consumer will accept. * @@ -281,7 +305,7 @@ export function packSyncBatchMessages(input: { * every message assert something false -- so the mismatch is possible and the * guard is not decorative. */ pruningKeys?: Readonly>; -}): SyncBatchMessage[] { +}): SyncBatchRowsMessage[] { const { lane, capturedAt, @@ -319,7 +343,7 @@ export function packSyncBatchMessages(input: { for (const row of rows) groups.push([row]); } - const messages: SyncBatchMessage[] = []; + const messages: SyncBatchRowsMessage[] = []; let current: Record[] = []; let currentBytes = 0; @@ -338,7 +362,7 @@ export function packSyncBatchMessages(input: { return rest; }) : current; - const message: SyncBatchMessage = { + const message: SyncBatchRowsMessage = { lane, captured_at: capturedAt, ...(passTotal !== undefined ? { pass_total: passTotal } : {}), @@ -706,10 +730,14 @@ export function passTallyFor( export function syncBatchRows( message: SyncBatchMessage, ): Record[] { - if (!message.key_column) return message.rows; + // A multi-family message has no `rows`, and never reaches here -- writeSyncBatch + // hands it to the family writer and returns. Empty rather than a throw so the + // fallback is the harmless one: a writer given nothing writes nothing. + const rows = message.rows ?? []; + if (!message.key_column) return rows; const column = message.key_column; const value = message.key_value; - return message.rows.map((row) => ({ ...row, [column]: value })); + return rows.map((row) => ({ ...row, [column]: value })); } /** How many rows a message actually carries, whichever shape it uses. */ diff --git a/tests/data-api-sync-queue-consumer.test.ts b/tests/data-api-sync-queue-consumer.test.ts index 4cacfb90e..ac0e98931 100644 --- a/tests/data-api-sync-queue-consumer.test.ts +++ b/tests/data-api-sync-queue-consumer.test.ts @@ -27,6 +27,10 @@ const SCHEMA = fs.readFileSync( path.join(process.cwd(), "migrations/d1/0020_account_balances_passes.sql"), "utf8", + ) + + fs.readFileSync( + path.join(process.cwd(), "migrations/d1/0010_chain_detail.sql"), + "utf8", ); const COLDKEY = "5CXRfP2ekFhYQ6BCwEy5V8YyxgLmUmTNzHZTKAfTHKhKPBqE"; @@ -212,4 +216,70 @@ describe("the sync queue consumer", () => { await consume([]); assert.equal(rows().length, 0); }); + + test("writes a multi-family message, and does not fail its batch-mates", async () => { + // THE REGRESSION (metagraphed-infra#359). The handler's opening log line + // summed `m.rows.length` over every valid message. A chain-detail message + // carries `families` and no `rows` at all, so that read threw a TypeError + // ABOVE the per-message try/catch -- taking the whole batch, every lane + // co-batched with it, five retries, into the dead-letter queue. Nothing + // caught it because `rows` was declared required on a shape that omits it. + // + // So this drives the families message through the REAL handler alongside a + // single-family neighbour, and asserts the neighbour survives. + const neighbour = positionMessage(); + const chainDetail = message({ + lane: "chain-detail", + captured_at: 1_780_000_000_000, + families: { + blockRows: [ + { + block_number: 6_100_000, + block_hash: `0x${"ab".repeat(32)}`, + spec_version: 441, + extrinsic_count: 1, + chain_event_count: 0, + account_event_count: 0, + observed_at: 1_780_000_000_000, + synced_at: 1_780_000_000_000, + }, + ], + extrinsicRows: [ + { + block_number: 6_100_000, + extrinsic_index: 0, + extrinsic_hash: `0x${"cd".repeat(32)}`, + signer: COLDKEY, + call_module: "SubtensorModule", + call_function: "set_weights", + success: 1, + fee_tao: 0, + observed_at: 1_780_000_000_000, + synced_at: 1_780_000_000_000, + }, + ], + }, + }); + + await consume([chainDetail, neighbour]); + + assert.deepEqual(chainDetail.calls, ["ack"]); + assert.deepEqual(neighbour.calls, ["ack"], "the batch-mate still landed"); + assert.equal(rows().length, 1, "and its rows were written"); + const blocks = db + .prepare("SELECT block_number FROM chain_detail_blocks") + .all() as Record[]; + assert.deepEqual( + blocks.map((row) => row.block_number), + [6_100_000], + ); + const extrinsics = db + .prepare("SELECT COUNT(*) AS n FROM chain_detail_extrinsics") + .all() as Record[]; + assert.equal( + extrinsics[0]!.n, + 1, + "both families landed, which is why they travel together", + ); + }); }); diff --git a/tests/sync-batch-queue.test.ts b/tests/sync-batch-queue.test.ts index db6185e94..c108209be 100644 --- a/tests/sync-batch-queue.test.ts +++ b/tests/sync-batch-queue.test.ts @@ -478,6 +478,19 @@ describe("packSyncBatchMessages", () => { assert.deepEqual(rebuilt[0], rows[0]); }); + test("a message with no rows reads as no rows, not as a crash", () => { + // A multi-family message has `families` and no `rows`, and never reaches + // here -- writeSyncBatch hands it to the family writer first. The fallback + // exists because the SAME missing field, read directly, crashed the + // consumer's batch log (metagraphed-infra#359): every read of `rows` now + // goes through a helper that survives its absence, and empty is the + // harmless answer -- a writer given nothing writes nothing. + assert.deepEqual( + syncBatchRows({ lane: "chain-detail", captured_at: 1, families: {} }), + [], + ); + }); + test("a hoisted key is refused if the rows still carry it", () => { // Two copies of one value can disagree, and the consumer would have to pick. assert.equal( diff --git a/workers/data-api.ts b/workers/data-api.ts index a47053b81..5ecaa01f8 100644 --- a/workers/data-api.ts +++ b/workers/data-api.ts @@ -420,6 +420,7 @@ import { enqueueSyncBatch, packMultiFamilyMessage, type SyncBatchFamilyWriters, + syncBatchRowCount, syncLaneUsesQueue, validSyncBatchMessage, writeSyncBatch, @@ -7291,7 +7292,12 @@ export default { ctx: ExecutionContext, ): Promise { const { valid, invalid } = classifySyncBatch(batch.messages); - const rows = valid.reduce((n: number, m) => n + m.rows.length, 0); + // syncBatchRowCount, not `m.rows.length`: a multi-family message carries + // `families` and no `rows` at all, so the direct read threw a TypeError HERE + // -- above the per-message try/catch below -- and failed the whole batch, + // every co-batched lane with it, five times over, into the dead-letter + // queue. A log line is the last thing that should be able to do that. + const rows = valid.reduce((n: number, m) => n + syncBatchRowCount(m), 0); const lanes = [...new Set(valid.map((m) => m.lane))].sort().join(","); console.log( `sync-batches: ${batch.messages.length} message(s), ${valid.length} valid ` +