Node.js、PostgreSQL 中的事务冲突、乐观并发控制和事务重试

Question

我想使用 PostgreSQL transaction isolation使用 optimistic concurrency control 确保数据正确性自动重试冲突事务的模式，而不是我的应用程序预先锁定数据库行和表。

实现此功能的一种常用方法是 Web 应用程序 retries the transaction代码块内的特定次数或 replays the HTTP request通过中间件层，也称为 HTTP 请求重放。 Here is an example of such a middleware for Pyramid and Python web applications web .

我没有找到任何关于 Node.js 及其 PostgreSQL 驱动程序如何处理有两个并发事务正在进行并且一个由于读写冲突而无法通过的情况的好的信息。 PostgreSQL 将回滚其中一个事务，但这是如何向应用程序发出信号的呢？在 Python 中，PSQL 驱动程序会引发 psycopg2.extensions.TransactionRollbackError在这种情况下。 For other SQL database drivers here are some exceptions they will raise .

当您将 SQL 事务隔离级别设置为 SERIALIZABLE 时，这种行为更为常见，因为您往往会在负载下遇到更多冲突，所以我想优雅地处理它，而不是向用户提供 HTTP 500。

我的问题是:

如果需要特殊处理并且重试库不能独立，如何使用 PostgreSQL 和一些常见的 ORM 框架(如 TypeORM)检测脏读回滚？

是否有中间件(NestJS/Express.js/others)来处理这个问题，并在数据库驱动程序发生事务回滚时自动尝试重播 HTTP 请求 N 次？

Answer 1

以下是使用 pg library 的库时如何处理并发性。比如 TypeORM:

/**
 * Check error code to determine if we should retry a transaction.
 * 
 * See https://www.postgresql.org/docs/10/errcodes-appendix.html and
 * https://stackoverflow.com/a/16409293/749644
 */
function shouldRetryTransaction(err: unknown) {
  const code = typeof err === 'object' ? String((err as any).code) : null
  return code === '40001' || code === '40P01';
}

/**
 * Using a repeatable read transaction throws an error with the code 40001 
 * "serialization failure due to concurrent update" if the user was 
 * updated by another concurrent transaction.
 */
async function updateUser(data: unknown) {
  try {
    return await this.userRepo.manager.transaction(
      'REPEATABLE READ',
      async manager => {
        const user = manager.findOne(User, id);
    
        // Modify user
        // ...
    
        // Save the user
        await manager.save(user);
      }
    );
  } catch (err) {
    if (shouldRetryTransaction(err)) {
      // retry logic     
    } else {
      throw err;
    }
  }
}

对于重试事务，我建议使用 async-retry 之类的库。它抽象了重试逻辑。
你会注意到这种模式非常适合简单的东西，但如果你想通过 manager周围(例如，事务可以在其他服务中重用)那么这将变得非常麻烦。我建议使用 typeorm-transactional-cls-hooked 利用持续本地存储来传播事务的库。
以下是如何重放 express 应用的交易:

/**
 * Request replay middleware
 */

import retry from 'async-retry';

function replayOnTransactionError(fn: (req, res, next) => unknown) {
  return (req, res, next) => {
    retry(bail => {
      try {
        // Call the actual handler
        await fn(req, res, next);
      } catch (err) {
        if (!shouldRetryTransaction(err)) {
          // Bail out if we're not supposed to retry anymore
          return bail(err);
        }

        // Rethrow error to continue retrying
        throw err;
      }
    }, {
      factor: 2,
      retries: 3,
      minTimeout: 30,
    });
  }
}

app.put('/users/:id', replayOnTransactionError(async (req, res, next) => {
  // ...
}))