概要

トランザクションを扱ったAzure Functions + SQL Serverの例があまりないので試してみたメモ。
ソースコード

コード

delete insertを例にした

apps/api/src/lib/database.ts


export default class Database {

+  private getPreparedStatement(transaction?: sql.Transaction) {
+    if (transaction) {
+      return new sql.PreparedStatement(transaction);
+    }
    if (this.poolconnection == null) {
      throw new Error('Database connection failed');
    }
    return new sql.PreparedStatement(this.poolconnection);
  }

  async executePreparedStatement<T>(
    args: ExecuteQueryArgs,
+    transaction?: sql.Transaction,
  ) {
    await this.connect();
    const ps = this.getPreparedStatement(transaction);
    const values = args.params.reduce((acc, { name, data, type }) => {
      ps.input(name, type);
      return { ...acc, [name]: data };
    }, {});

    try {
      await ps.prepare(args.query);
      const result = await ps.execute<T>(values);
      return result;
    } catch (e) {
      args.logger(e);
      throw e;
    } finally {
      await ps.unprepare();
    }
  }
+  private getRequest(transaction?: sql.Transaction) {
+    if (transaction) {
+      return new sql.Request(transaction);
+    }
    if (this.poolconnection == null) {
      throw new Error('Database connection failed');
    }
    return new sql.Request(this.poolconnection);
  }
  async bulk(
    args: {
      table: sql.Table;
      // eslint-disable-next-line @typescript-eslint/no-explicit-any
      logger: (...args: any[]) => void;
    },
+    transaction?: sql.Transaction,
  ) {
    await this.connect();
    const request = this.getRequest(transaction);
    const result = await request.bulk(args.table);
    return result;
  }
+  private async createTransaction() {
+    await this.connect();
+    if (this.poolconnection == null) {
+      throw new Error('Database connection failed');
+    }
+    const transaction = new sql.Transaction(this.poolconnection);
+    return transaction;
+  }

+  async useTransaction(
+    callback: (transaction: sql.Transaction) => Promise<void>,
+  ) {
+    const transaction = await this.createTransaction();
+    try {
+      await transaction.begin();
+      await callback(transaction);
+      await transaction.commit();
+    } catch (e) {
+      await transaction.rollback();
+      throw e;
+    }
+  }
}

apps/api/src/routes/transactionTest.ts

import { Hono } from 'hono';
import { zValidator } from '@hono/zod-validator';
import { TransactionTestSchema } from '@db/zod';
import Database from '@api/lib/database';
import * as sql from 'mssql';

// eslint-disable-next-line turbo/no-undeclared-env-vars
const connectionString = process.env.CONNECTION_STRING!;
const db = new Database(connectionString); // 接続を使いまわすため、関数の外で定義
console.log('connection', connectionString);
const app = new Hono()
  .post(
    '/',
    zValidator('json', TransactionTestSchema.array().min(1)),
    async (c) => {
      const data = await c.req.valid('json');
      await db.useTransaction(async (transaction) => {
        const [{ TestID }] = data;
        const deleteArgs = {
          logger: console.log,
          query: 'DELETE from atrpg.TransactionTest where TestID = @TestID',
          params: [{ name: 'TestID', data: TestID, type: sql.NVarChar }],
        };

        const table = new sql.Table('atrpg.TransactionTest');
        // prettier-ignore
        table.columns.add('TestID', sql.NVarChar, { nullable: false, primary: true });
        // prettier-ignore
        table.columns.add('CreatedAt', sql.DateTimeOffset, { nullable: false, primary: true });
        data.forEach((row) => {
          table.rows.add(row.TestID, row.CreatedAt);
        });

        await db.executePreparedStatement(deleteArgs, transaction);
        await db.bulk({ table, logger: console.log }, transaction);
      });
      return c.json({});
    },
  );

export default app;

しょーもないエラー

[2024-08-15T13:16:11.613Z] TransactionError: Can't rollback transaction. There is a request in progress.

await漏れ。

  async executePreparedStatement<T>(args: ExecuteQueryArgs) {
    await this.connect();
    const ps = this.getPreparedStatement();
    try {
      const values = args.params.reduce((acc, { name, data, type }) => {
        ps.input(name, type);
        return { ...acc, [name]: data };
      }, {});
      await ps.prepare(args.query);
      const result = await ps.execute<T>(values);
      return result;
    } finally {
-    ps.unprepare();
+    await ps.unprepare();
    }
  }

タイムアウトエラー

このときのソース

下記のように時間のかかる処理があった場合にコネクションが迷子になる。

  .post(
    '/delay',
    zValidator('json', TransactionTestSchema.array().min(1)),
    async (c) => {
      const data = await c.req.valid('json');
      await db.beginTransaction();
      try {
        const [{ TestID }] = data;
        const deleteArgs = {
          logger: console.log,
          query: 'DELETE from atrpg.TransactionTest where TestID = @TestID',
          params: [{ name: 'TestID', data: TestID, type: sql.NVarChar }],
        };

        const table = new sql.Table('atrpg.TransactionTest');
        // prettier-ignore
        table.columns.add('TestID', sql.NVarChar, { nullable: false, primary: true });
        // prettier-ignore
        table.columns.add('CreatedAt', sql.DateTimeOffset, { nullable: false, primary: true });
        data.forEach((row) => {
          table.rows.add(row.TestID, row.CreatedAt);
        });

        await db.executePreparedStatement(deleteArgs);
        await new Promise((resolve) => setTimeout(resolve, 1000 * 5));
        await db.bulk({ table, logger: console.log });
        await db.commitTransaction();
        return c.json({});
      } catch (e) {
        await db.rollbackTransaction();
        throw e;
      }
    },
  );

###
post http://localhost:7071/api/test/delay
Content-Type: application/json

[{"TestID": "4", "CreatedAt": "2022-01-04T09:00:00+09:00"}]

###
post http://localhost:7071/api/test
Content-Type: application/json

[{"TestID": "1", "CreatedAt": "2022-01-01T09:00:00+09:00"}]

[2024-08-15T13:54:42.962Z] RequestError: Timeout: Request failed to complete in 15000ms

post http://localhost:7071/api/test
Content-Type: application/json

[{"TestID": "1", "CreatedAt": "2022-01-01T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-02T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-03T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-04T09:00:00+09:00"}
]

###
post http://localhost:7071/api/test
Content-Type: application/json

[{"TestID": "2", "CreatedAt": "2022-01-01T09:00:00+09:00"},
{"TestID": "3", "CreatedAt": "2022-01-02T09:00:00+09:00"},
{"TestID": "4", "CreatedAt": "2022-01-03T09:00:00+09:00"}
]

###
post http://localhost:7071/api/test
Content-Type: application/json

[
{"TestID": "2", "CreatedAt": "2022-01-02T09:00:00+09:00"},
{"TestID": "2", "CreatedAt": "2022-01-02T09:00:00+09:00"}
]


###
post http://localhost:7071/api/test/delay
Content-Type: application/json

[
{"TestID": "3", "CreatedAt": "2022-01-02T09:00:00+09:00"},
{"TestID": "3", "CreatedAt": "2022-01-02T09:00:00+09:00"}
]

###
post http://localhost:7071/api/test/delay
Content-Type: application/json

[
{"TestID": "4", "CreatedAt": "2022-01-04T09:00:00+09:00"}]

###
post http://localhost:7071/api/test
Content-Type: application/json

[{"TestID": "1", "CreatedAt": "2022-01-01T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-02T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-03T09:00:00+09:00"},
{"TestID": "1", "CreatedAt": "2022-01-04T09:00:00+09:00"}
]

下記のように、トランザクションはリクエストのスコープとし、使いまわさないようにすると上記の減少は解消された。

  .post(
    '/delay',
    zValidator('json', TransactionTestSchema.array().min(1)),
    async (c) => {
      const data = await c.req.valid('json');
      const transaction = await db.beginTransaction();
      try {
        const [{ TestID }] = data;
        const deleteArgs = {
          logger: console.log,
          query: 'DELETE from atrpg.TransactionTest where TestID = @TestID',
          params: [{ name: 'TestID', data: TestID, type: sql.NVarChar }],
        };

        const table = new sql.Table('atrpg.TransactionTest');
        // prettier-ignore
        table.columns.add('TestID', sql.NVarChar, { nullable: false, primary: true });
        // prettier-ignore
        table.columns.add('CreatedAt', sql.DateTimeOffset, { nullable: false, primary: true });
        data.forEach((row) => {
          table.rows.add(row.TestID, row.CreatedAt);
        });

        await db.executePreparedStatement(deleteArgs, transaction);
        await new Promise((resolve) => setTimeout(resolve, 1000 * 5));
        await db.bulk({ table, logger: console.log }, transaction);
        await transaction.commit();
        return c.json({});
      } catch (e) {
        await transaction.rollback();
        throw e;
      }
    },
  );

SQL Serverのログを見る

成功しているときのログ。

失敗したときのログ。

TSQLをみると、DEADLOCKの文字があるが、これは単純にSETしているだけで意味がなさそう。

package0.event_fileログは下記。成功した200の接続のログしか出ないようで、あまりいいデータは見つけられなかった。

data	<system spinlockBackoffs="0" sickSpinlockType="none" sickSpinlockTypeAfterAv="none" latchWarnings="0" isAccessViolationOccurred="0" writeAccessViolationCount="0" totalDumpRequests="0" intervalDumpRequests="0" nonYieldingTasksReported="0" pageFaults="0" systemCpuUtilization="100" sqlCpuUtilization="0" BadPagesDetected="0" BadPagesFixed="0" LastBadPageAddress="0x0"/>
data	<resource lastNotification="RESOURCE_MEMPHYSICAL_HIGH" outOfMemoryExceptions="0" isAnyPoolOutOfMemory="0" processOutOfMemoryPeriod="0"><memoryReport name="Process/System Counts" unit="Value"><entry description="Available Physical Memory" value="25143144448"/><entry description="Available Virtual Memory" value="70364449144833"/><entry description="Available Paging File" value="25143144448"/><entry description="Working Set" value="4294967296"/><entry description="Percent of Committed Memory in WS" value="100"/><entry description="Page Faults" value="0"/><entry description="System physical memory high" value="1"/><entry description="System physical memory low" value="0"/><entry description="Process physical memory low" value="0"/><entry description="Process virtual memory low" value="0"/></memoryReport><memoryReport name="Memory Manager" unit="KB"><entry description="VM Reserved" value="36698476"/><entry description="VM Committed" value="579432"/><entry description="Locked Pages Allocated" value="0"/><entry description="Large Pages Allocated" value="135168"/><entry description="Emergency Memory" value="1024"/><entry description="Emergency Memory In Use" value="8"/><entry description="Target Committed" value="23925832"/><entry description="Current Committed" value="579432"/><entry description="Pages Allocated" value="352328"/><entry description="Pages Reserved" value="0"/><entry description="Pages Free" value="14176"/><entry description="Pages In Use" value="512528"/><entry description="Page Alloc Potential" value="23301832"/><entry description="NUMA Growth Phase" value="0"/><entry description="Last OOM Factor" value="0"/><entry description="Last OS Error" value="0"/></memoryReport></resource>
data	<queryProcessing maxWorkers="832" workersCreated="155" workersIdle="86" tasksCompletedWithinInterval="1488" pendingTasks="0" oldestPendingTaskWaitingTime="0" hasUnresolvableDeadlockOccurred="0" hasDeadlockedSchedulersOccurred="0" trackingNonYieldingScheduler="0x0"><topWaits><nonPreemptive><byCount><wait waitType="MEMORY_ALLOCATION_EXT" waits="111874" averageWaitTime="0" maxWaitTime="7"/><wait waitType="RESERVED_MEMORY_ALLOCATION_EXT" waits="3455" averageWaitTime="0" maxWaitTime="0"/><wait waitType="HADR_FILESTREAM_IOMGR_IOCOMPLETION" waits="1200" averageWaitTime="499" maxWaitTime="501"/><wait waitType="WRITELOG" waits="629" averageWaitTime="0" maxWaitTime="1"/><wait waitType="PAGEIOLATCH_UP" waits="420" averageWaitTime="0" maxWaitTime="1"/><wait waitType="PAGEIOLATCH_SH" waits="380" averageWaitTime="0" maxWaitTime="3"/><wait waitType="PAGELATCH_SH" waits="366" averageWaitTime="0" maxWaitTime="2"/><wait waitType="PAGELATCH_EX" waits="366" averageWaitTime="0" maxWaitTime="0"/><wait waitType="WRITE_COMPLETION" waits="290" averageWaitTime="0" maxWaitTime="2"/><wait waitType="IO_COMPLETION" waits="239" averageWaitTime="0" maxWaitTime="1"/></byCount><byDuration><wait waitType="DISPATCHER_QUEUE_SEMAPHORE" waits="90" averageWaitTime="56286" maxWaitTime="120001"/><wait waitType="HADR_FILESTREAM_IOMGR_IOCOMPLETION" waits="1200" averageWaitTime="499" maxWaitTime="501"/><wait waitType="PWAIT_EXTENSIBILITY_CLEANUP_TASK" waits="3" averageWaitTime="120000" maxWaitTime="300000"/><wait waitType="LCK_M_U" waits="50" averageWaitTime="488" maxWaitTime="15012"/><wait waitType="LCK_M_X" waits="6" averageWaitTime="2512" maxWaitTime="15005"/><wait waitType="LATCH_EX" waits="11" averageWaitTime="967" maxWaitTime="3533"/><wait waitType="LCK_M_S" waits="16" averageWaitTime="511" maxWaitTime="3538"/><wait waitType="LCK_M_SCH_S" waits="2" averageWaitTime="1755" maxWaitTime="3466"/><wait waitType="LCK_M_SCH_M" waits="3" averageWaitTime="1156" maxWaitTime="3469"/><wait waitType="PWAIT_ALL_COMPONENTS_INITIALIZED" waits="5" averageWaitTime="423" maxWaitTime="508"/></byDuration></nonPreemptive><preemptive><byCount><wait waitType="PREEMPTIVE_OS_FLUSHFILEBUFFERS" waits="1261" averageWaitTime="1" maxWaitTime="25"/><wait waitType="PREEMPTIVE_OS_QUERYREGISTRY" waits="968" averageWaitTime="0" maxWaitTime="0"/><wait waitType="PREEMPTIVE_OS_FILEOPS" waits="483" averageWaitTime="0" maxWaitTime="1"/><wait waitType="PREEMPTIVE_XE_CALLBACKEXECUTE" waits="421" averageWaitTime="0" maxWaitTime="0"/><wait waitType="PREEMPTIVE_OS_CREATEFILE" waits="116" averageWaitTime="1" maxWaitTime="90"/><wait waitType="PREEMPTIVE_OS_AUTHENTICATIONOPS" waits="111" averageWaitTime="0" maxWaitTime="3"/><wait waitType="PREEMPTIVE_OS_CRYPTOPS" waits="109" averageWaitTime="5" maxWaitTime="638"/><wait waitType="PREEMPTIVE_OS_AUTHORIZATIONOPS" waits="106" averageWaitTime="0" maxWaitTime="0"/><wait waitType="PREEMPTIVE_OS_GETFILEATTRIBUTES" waits="76" averageWaitTime="0" maxWaitTime="0"/><wait waitType="PREEMPTIVE_OS_CRYPTACQUIRECONTEXT" waits="68" averageWaitTime="0" maxWaitTime="0"/></byCount><byDuration><wait waitType="PREEMPTIVE_OS_FLUSHFILEBUFFERS" waits="1261" averageWaitTime="1" maxWaitTime="25"/><wait waitType="PREEMPTIVE_OS_CRYPTOPS" waits="109" averageWaitTime="5" maxWaitTime="638"/><wait waitType="PREEMPTIVE_OS_QUERYREGISTRY" waits="968" averageWaitTime="0" maxWaitTime="0"/><wait waitType="PREEMPTIVE_OS_FILEOPS" waits="483" averageWaitTime="0" maxWaitTime="1"/><wait waitType="PREEMPTIVE_OS_CREATEFILE" waits="116" averageWaitTime="1" maxWaitTime="90"/><wait waitType="PREEMPTIVE_OS_WRITEFILE" waits="30" averageWaitTime="3" maxWaitTime="10"/><wait waitType="PREEMPTIVE_OS_AUTHENTICATIONOPS" waits="111" averageWaitTime="0" maxWaitTime="3"/><wait waitType="PREEMPTIVE_OS_WRITEFILEGATHER" waits="56" averageWaitTime="0" maxWaitTime="18"/><wait waitType="PREEMPTIVE_XE_TARGETINIT" waits="5" averageWaitTime="5" maxWaitTime="27"/><wait waitType="PREEMPTIVE_OS_VERIFYTRUST" waits="4" averageWaitTime="6" maxWaitTime="8"/></byDuration></preemptive></topWaits><cpuIntensiveRequests></cpuIntensiveRequests><pendingTasks></pendingTasks><blockingTasks></blockingTasks></queryProcessing>
data	<ioSubsystem ioLatchTimeouts="0" intervalLongIos="0" totalLongIos="0"><longestPendingRequests></longestPendingRequests></ioSubsystem>

参考

SqlServerにmssqlを使って接続してJsonの入ったカラムから値を取り出したメモ