Add per-row monotonic tie-breaker to enqueueMany()

MySQL evaluates NOW(6) once per statement, so all rows in a bulk INSERT
share the same deliver_after timestamp.  When messages share an
orderingKey, dequeueing (which ORDER BY deliver_after) then selects among
them non-deterministically.

Fix by using delayMs * 1000 + index microseconds for each row's interval,
so row i gets a deliver_after exactly i µs later than row 0.  This
ensures the insertion order is preserved under any orderingKey without
requiring a new column or schema change.

Add a regression test that enqueues five messages under the same ordering
key via enqueueMany() and asserts they are delivered in order.

#599 (comment)

Co-Authored-By: claude-sonnet-4-6 <claude-sonnet-4-6@anthropic.com>

Author: dahlia

packages/mysql/src/mq.test.ts

@@ -575,6 +575,48 @@ test(
   },
 );
 
+test(
+  "MysqlMessageQueue.enqueueMany() preserves insertion order for same ordering key",
+  { skip: dbUrl == null },
+  async () => {
+    if (dbUrl == null) return; // Bun does not support skip option
+    const pool = mysql.createPool(dbUrl!);
+    const tableName = randomTableName("bord");
+    const mq = new MysqlMessageQueue(pool, {
+      tableName,
+      pollInterval: { milliseconds: 50 },
+    });
+    const controller = new AbortController();
+    const received: number[] = [];
+    try {
+      const listening = mq.listen(
+        (msg: number) => {
+          received.push(msg);
+        },
+        { signal: controller.signal },
+      );
+
+      // All 5 messages share the same ordering key.  Without per-row
+      // tie-breaker timestamps, all deliver_after values would be identical
+      // (NOW(6) is constant within a single SQL statement) and dequeue order
+      // would be nondeterministic.
+      await mq.enqueueMany([1, 2, 3, 4, 5], { orderingKey: "order-test" });
+
+      await waitFor(() => received.length >= 5, 10_000);
+      assert.deepStrictEqual(
+        received,
+        [1, 2, 3, 4, 5],
+        "enqueueMany() must deliver messages in insertion order for the same ordering key",
+      );
+      controller.abort();
+      await listening;
+    } finally {
+      await mq.drop();
+      await pool.end();
+    }
+  },
+);
+
 // ---------------------------------------------------------------------------
 // Handler error survival
 // ---------------------------------------------------------------------------

packages/mysql/src/mq.ts

@@ -245,9 +245,14 @@ export class MysqlMessageQueue implements MessageQueue {
     const placeholders = messages.map(() =>
       "(UUID(), CAST(? AS JSON), DATE_ADD(NOW(6), INTERVAL ? MICROSECOND), ?)"
     ).join(", ");
-    const values = messages.flatMap((message) => [
+    // Each row gets a monotonically increasing interval so that messages
+    // sharing the same orderingKey are assigned distinct deliver_after values
+    // within a single INSERT statement.  Without this tie-breaker, all rows
+    // would receive the same NOW(6) timestamp (MySQL evaluates NOW(6) once per
+    // statement), making dequeue order nondeterministic for ordered keys.
+    const values = messages.flatMap((message, index) => [
       JSON.stringify(message),
-      delayMs * 1000,
+      delayMs * 1000 + index,
       orderingKey,
     ]);
     let conn: PoolConnection | undefined;