fix(queue): count a multi-family message's rows without crashing its batch

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<string, unknown>[];
+  /**
+   * 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<string, unknown>[];
   /**
    * 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<string, unknown>[];
+};
+
 /**
  * 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<Record<string, string>>;
-}): 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<string, unknown>[] = [];
   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<string, unknown>[] {
-  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. */

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<string, unknown>[];
+    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<string, unknown>[];
+    assert.equal(
+      extrinsics[0]!.n,
+      1,
+      "both families landed, which is why they travel together",
+    );
+  });
 });

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(

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<void> {
     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 ` +